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 source, target, err := m.checkBranches(ctx, name, params, &v)
192 if err != nil {
193 return domain.MergeRequest{}, err
194 }
195 if err := v.Err(); err != nil {
196 return domain.MergeRequest{}, err
197 }
198
199 // Under the lock, so that two requests for the same pair cannot both pass
200 // the check for an existing one.
201 unlock, err := m.lockRepo(ctx, name)
202 if err != nil {
203 return domain.MergeRequest{}, err
204 }
205 defer unlock()
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 d.Commits, d.CommitsTruncated, err = m.git.ListRange(ctx, name, exclude, head, MaxMergeRequestCommits); err != nil {
418 return MergeRequestDetail{}, err
419 }
420 if sinceLastRead && !d.LastReadID.IsZero() {
421 switch ok, err := m.git.HasCommit(ctx, name, d.LastReadID); {
422 case err != nil:
423 return MergeRequestDetail{}, err
424 case ok:
425 base = d.LastReadID
426 default:
427 d.LastReadLost = true
428 }
429 }
430 if base.IsZero() {
431 return d, nil
432 }
433 d.DiffBaseID = base
434 if d.Stats, d.Diff, err = m.git.DiffCommits(ctx, name, base, head, domain.DefaultDiffLimits()); err != nil {
435 return MergeRequestDetail{}, err
436 }
437 return d, nil
438}
439
440// rangeOf returns what the review of a request compares: the commit whose
441// history the request's commits exclude, and the base of its diff. For an open
442// request they are the tip of the target and the merge base of the target and
443// the head; both are zero when the target is gone or shares no history with
444// the head. A merged or closed request keeps the base it stored for both.
445func (m *MergeRequests) rangeOf(ctx context.Context, mr domain.MergeRequest) (exclude, base domain.ObjectID, err error) {
446 if !mr.IsOpen() {
447 return mr.BaseID, mr.BaseID, nil
448 }
449 if mr.HeadID.IsZero() {
450 return domain.ObjectID{}, domain.ObjectID{}, nil
451 }
452 tip, err := m.branchOrZero(ctx, mr.Repo, mr.Target)
453 if err != nil || tip.IsZero() {
454 return domain.ObjectID{}, domain.ObjectID{}, err
455 }
456 base, err = m.git.MergeBase(ctx, mr.Repo, tip, mr.HeadID)
457 if errors.Is(err, domain.ErrNotFound) {
458 return tip, domain.ObjectID{}, nil
459 }
460 return tip, base, err
461}
462
463// AddCommentParams describes the comment to add.
464type AddCommentParams struct {
465 // Path, Line and Side designate a line of the diff; all empty for a
466 // general comment.
467 Path string
468 Line int
469 Side domain.DiffSide
470 Text string
471}
472
473// AddComment adds a comment and publishes CommentAdded. The principal needs
474// mr:write. The text must follow the rules of ADR 0017 and not be blank; a
475// comment on a line must name a line of the current diff of the request (see
476// domain.Diff.HasLine), else the call fails with domain.ErrInvalid. A request
477// that is not open gives domain.ErrPreconditionFailed, and one that already
478// holds MaxCommentsPerMergeRequest comments domain.ErrResourceExhausted.
479func (m *MergeRequests) AddComment(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, params AddCommentParams) (domain.Comment, error) {
480 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
481 if err != nil {
482 return domain.Comment{}, err
483 }
484 mr, err := m.store.Get(ctx, name, number)
485 if err != nil {
486 return domain.Comment{}, err
487 }
488 var v domain.Violations
489 text, err := domain.ParsePlainText(params.Text)
490 switch {
491 case err != nil:
492 v.AddErr("text", err)
493 case text.IsBlank():
494 v.Add("text", "is required")
495 }
496 general := params.Path == ""
497 switch {
498 case general && (params.Line != 0 || params.Side != domain.SideNone):
499 v.Add("path", "is required with line and side")
500 case !general && params.Line < 1:
501 v.Add("line", "must be 1 or more with a path")
502 case !general && params.Side != domain.SideOld && params.Side != domain.SideNew:
503 v.Add("side", "is required with a path")
504 }
505 if err := v.Err(); err != nil {
506 return domain.Comment{}, err
507 }
508 if !mr.IsOpen() {
509 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
510 }
511
512 if mr, err = m.withHead(ctx, mr); err != nil {
513 return domain.Comment{}, err
514 }
515 if !general {
516 if err := m.checkLine(ctx, mr, params); err != nil {
517 return domain.Comment{}, err
518 }
519 }
520
521 var c domain.Comment
522 for range maxWriteRetries {
523 if mr.CommentSeq >= MaxCommentsPerMergeRequest {
524 return domain.Comment{}, fmt.Errorf("%w: !%d already has %d comments", domain.ErrResourceExhausted, number, MaxCommentsPerMergeRequest)
525 }
526 c = domain.Comment{
527 ID: mr.CommentSeq + 1, Author: authorOf(p), Path: params.Path, Line: params.Line, Side: params.Side,
528 CommitID: mr.HeadID, Text: text, CreatedAt: m.clock.Now().UTC(),
529 }
530 mr.CommentSeq, mr.UpdatedAt = c.ID, c.CreatedAt
531 err = m.store.SaveComment(ctx, mr, c)
532 if !errors.Is(err, domain.ErrConflict) {
533 break
534 }
535 // Another write got in between: read the request again.
536 fresh, gerr := m.store.Get(ctx, name, number)
537 if gerr != nil {
538 return domain.Comment{}, gerr
539 }
540 if !fresh.IsOpen() {
541 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, fresh.State)
542 }
543 fresh.HeadID = mr.HeadID
544 mr = fresh
545 }
546 if err != nil {
547 return domain.Comment{}, fmt.Errorf("save comment on !%d of %q: %w", number, name, err)
548 }
549 log := zerowrap.FromCtx(ctx)
550 log.Info().Str("repo", name.String()).Int64("number", number).Int64("comment", c.ID).
551 Bool("on_line", !general).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("comment added")
552 m.bus.Publish(ctx, domain.CommentAdded{
553 Repo: name, Number: number, Comment: c, At: c.CreatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
554 })
555 return c, nil
556}
557
558// checkLine verifies that a comment designates a line of the current diff of
559// the request.
560func (m *MergeRequests) checkLine(ctx context.Context, mr domain.MergeRequest, params AddCommentParams) error {
561 var v domain.Violations
562 _, base, err := m.rangeOf(ctx, mr)
563 if err != nil {
564 return err
565 }
566 if mr.HeadID.IsZero() || base.IsZero() {
567 v.Add("path", "the merge request has no diff to comment on")
568 return v.Err()
569 }
570 _, diff, err := m.git.DiffCommits(ctx, mr.Repo, base, mr.HeadID, domain.DefaultDiffLimits())
571 if err != nil {
572 return err
573 }
574 if !diff.HasLine(params.Path, params.Side, params.Line) {
575 v.Add("path", fmt.Sprintf("%s:%d on the %s side is not a line of the diff", params.Path, params.Line, params.Side))
576 }
577 return v.Err()
578}
579
580// ResolveComment marks a comment resolved or unresolved. The principal needs
581// mr:write. An unknown comment gives domain.ErrNotFound and a request that is
582// not open domain.ErrPreconditionFailed. Setting the state the comment already
583// has succeeds and writes nothing.
584func (m *MergeRequests) ResolveComment(ctx context.Context, p domain.Principal, name domain.RepoName, number, commentID int64, resolved bool) (domain.Comment, error) {
585 if _, err := m.writable(ctx, p, name, domain.ScopeMRWrite); err != nil {
586 return domain.Comment{}, err
587 }
588 for range maxWriteRetries {
589 mr, err := m.store.Get(ctx, name, number)
590 if err != nil {
591 return domain.Comment{}, err
592 }
593 c, err := m.store.GetComment(ctx, name, number, commentID)
594 if err != nil {
595 return domain.Comment{}, err
596 }
597 if !mr.IsOpen() {
598 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
599 }
600 if c.Resolved == resolved {
601 return c, nil
602 }
603 c.Resolved = resolved
604 mr.UpdatedAt = m.clock.Now().UTC()
605 err = m.store.SaveComment(ctx, mr, c)
606 if errors.Is(err, domain.ErrConflict) {
607 continue
608 }
609 if err != nil {
610 return domain.Comment{}, fmt.Errorf("resolve comment %d of !%d in %q: %w", commentID, number, name, err)
611 }
612 return c, nil
613 }
614 return domain.Comment{}, fmt.Errorf("resolve comment %d of !%d in %q: %w", commentID, number, name, domain.ErrConflict)
615}
616
617// Close closes an open merge request without merging it and publishes
618// MergeRequestClosed. The principal needs mr:write. A request that is not open
619// gives domain.ErrPreconditionFailed. The request keeps the head it had and
620// its merge base, so that its diff stays readable.
621func (m *MergeRequests) Close(ctx context.Context, p domain.Principal, name domain.RepoName, number int64) (domain.MergeRequest, error) {
622 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
623 if err != nil {
624 return domain.MergeRequest{}, err
625 }
626 unlock, err := m.lockRepo(ctx, name) // a merge and a close of one request exclude each other
627 if err != nil {
628 return domain.MergeRequest{}, err
629 }
630 defer unlock()
631 for range maxWriteRetries {
632 mr, err := m.store.Get(ctx, name, number)
633 if err != nil {
634 return domain.MergeRequest{}, err
635 }
636 if !mr.IsOpen() {
637 return domain.MergeRequest{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
638 }
639 if mr, err = m.withHead(ctx, mr); err != nil {
640 return domain.MergeRequest{}, err
641 }
642 _, base, err := m.rangeOf(ctx, mr)
643 if err != nil {
644 return domain.MergeRequest{}, err
645 }
646 mr.State, mr.BaseID, mr.UpdatedAt = domain.MRStateClosed, base, m.clock.Now().UTC()
647 err = m.store.Update(ctx, mr)
648 if errors.Is(err, domain.ErrConflict) {
649 continue
650 }
651 if err != nil {
652 return domain.MergeRequest{}, fmt.Errorf("close !%d of %q: %w", number, name, err)
653 }
654 log := zerowrap.FromCtx(ctx)
655 log.Info().Str("repo", name.String()).Int64("number", number).
656 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request closed")
657 m.bus.Publish(ctx, domain.MergeRequestClosed{MergeRequest: mr, At: mr.UpdatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
658 return mr, nil
659 }
660 return domain.MergeRequest{}, fmt.Errorf("close !%d of %q: %w", number, name, domain.ErrConflict)
661}
662
663// MarkRead records the head the reviewer has read, for the since-last-read diff
664// of Get. The principal needs mr:write. A zero head means the current head of
665// the source branch (the stored one for a merged or closed request); one that
666// is not a commit of the repository gives domain.ErrInvalid. It returns the
667// head recorded.
668func (m *MergeRequests) MarkRead(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, head domain.ObjectID) (domain.ObjectID, error) {
669 if _, err := m.writable(ctx, p, name, domain.ScopeMRWrite); err != nil {
670 return domain.ObjectID{}, err
671 }
672 mr, err := m.store.Get(ctx, name, number)
673 if err != nil {
674 return domain.ObjectID{}, err
675 }
676 if head.IsZero() {
677 if mr, err = m.withHead(ctx, mr); err != nil {
678 return domain.ObjectID{}, err
679 }
680 if head = mr.HeadID; head.IsZero() {
681 return domain.ObjectID{}, fmt.Errorf("%w: the source branch %q no longer exists, give a head", domain.ErrPreconditionFailed, mr.Source)
682 }
683 } else {
684 var v domain.Violations
685 switch ok, err := m.git.HasCommit(ctx, name, head); {
686 case err != nil:
687 return domain.ObjectID{}, err
688 case !ok:
689 v.Add("head_id", fmt.Sprintf("%s is not a commit of the repository", head))
690 return domain.ObjectID{}, v.Err()
691 }
692 }
693 if err := m.store.SetRead(ctx, name, number, head); err != nil {
694 return domain.ObjectID{}, err
695 }
696 return head, nil
697}
698
699// MergeParams describes the merge to perform.
700type MergeParams struct {
701 // Mode defaults to fast-forward.
702 Mode domain.MergeMode
703 // HeadID is the commit the caller reviewed; the source branch must still
704 // point to it.
705 HeadID domain.ObjectID
706}
707
708// Merge merges an open merge request, publishes MergeRequestMerged then a
709// PushEvent for the target branch, and returns the merged request. The
710// principal needs mr:merge. Merges of one repository run one at a time.
711//
712// It fails with domain.ErrInvalid for a zero head, and with
713// domain.ErrPreconditionFailed when the request is not open, the mode is not
714// allowed by the settings of the repository, the source branch no longer
715// points to the head ("branch moved, read again") or is gone, the target is
716// gone or is not an ancestor of the head (the author rebases and pushes
717// again), or the branch brings nothing. Checks come with CI (CI-1): there are
718// none yet, so the setting that lets failing checks block the merge has
719// nothing to act on.
720//
721// A fast-forward moves the target to the head, with the old value as a
722// precondition. A squash writes one commit with the tree of the head, the
723// target as parent and the author and message of domain.NewSquashCommit,
724// signed by the instance key, and moves the target to it the same way.
725func (m *MergeRequests) Merge(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, params MergeParams) (domain.MergeRequest, error) {
726 repo, err := m.writable(ctx, p, name, domain.ScopeMRMerge)
727 if err != nil {
728 return domain.MergeRequest{}, err
729 }
730 mode := params.Mode
731 if mode == "" {
732 mode = domain.MergeModeFastForward
733 }
734 var v domain.Violations
735 if _, err := domain.ParseMergeMode(mode.String()); err != nil {
736 v.Add("mode", "is not a merge mode")
737 }
738 if params.HeadID.IsZero() {
739 v.Add("head_id", "is required")
740 }
741 if err := v.Err(); err != nil {
742 return domain.MergeRequest{}, err
743 }
744
745 unlock, err := m.lockRepo(ctx, name)
746 if err != nil {
747 return domain.MergeRequest{}, err
748 }
749 defer unlock()
750
751 mr, err := m.store.Get(ctx, name, number)
752 if err != nil {
753 return domain.MergeRequest{}, err
754 }
755 if !mr.IsOpen() {
756 return domain.MergeRequest{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
757 }
758 // The settings are read again under the lock: a change applies at once.
759 repo, err = m.meta.Get(ctx, name)
760 if err != nil {
761 if errors.Is(err, domain.ErrNotFound) {
762 return domain.MergeRequest{}, err // deleted meanwhile
763 }
764 return domain.MergeRequest{}, fmt.Errorf("get repository %q: %w", name, err)
765 }
766 settings := repo.Settings
767 if settings.IsZero() {
768 settings = domain.DefaultRepoSettings()
769 }
770 if !settings.Allows(mode) {
771 return domain.MergeRequest{}, fmt.Errorf("%w: the %s merge mode is not allowed in %s", domain.ErrPreconditionFailed, mode, name)
772 }
773
774 head, err := m.branchOrZero(ctx, name, mr.Source)
775 switch {
776 case err != nil:
777 return domain.MergeRequest{}, err
778 case head.IsZero():
779 return domain.MergeRequest{}, fmt.Errorf("%w: the source branch %q no longer exists", domain.ErrPreconditionFailed, mr.Source)
780 case head != params.HeadID:
781 return domain.MergeRequest{}, fmt.Errorf("%w: branch moved, read again: %q is at %s, not %s", domain.ErrPreconditionFailed, mr.Source, head, params.HeadID)
782 }
783 tip, err := m.branchOrZero(ctx, name, mr.Target)
784 switch {
785 case err != nil:
786 return domain.MergeRequest{}, err
787 case tip.IsZero():
788 return domain.MergeRequest{}, fmt.Errorf("%w: the target branch %q no longer exists", domain.ErrPreconditionFailed, mr.Target)
789 case tip == head:
790 return domain.MergeRequest{}, fmt.Errorf("%w: nothing to merge, %q already is at %s", domain.ErrPreconditionFailed, mr.Target, head)
791 }
792 switch ok, err := m.git.IsAncestor(ctx, name, tip, head); {
793 case err != nil:
794 return domain.MergeRequest{}, err
795 case !ok:
796 return domain.MergeRequest{}, 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)
797 }
798
799 merged, err := m.moveTarget(ctx, mr, mode, tip, head)
800 if err != nil {
801 return domain.MergeRequest{}, err
802 }
803 m.recordPush(ctx, name) // the target moved, whatever happens next
804 mr, err = m.markMerged(ctx, name, number, mode, tip, head, merged)
805 if err != nil {
806 return domain.MergeRequest{}, err
807 }
808 log := zerowrap.FromCtx(ctx)
809 log.Info().Str("repo", name.String()).Int64("number", number).Str("mode", mode.String()).
810 Str("target", mr.Target).Str("commit", merged.String()).
811 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request merged")
812 at := mr.UpdatedAt
813 m.bus.Publish(ctx, domain.MergeRequestMerged{MergeRequest: mr, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
814 m.bus.Publish(ctx, domain.PushEvent{
815 Repo: name, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
816 Updates: []domain.RefUpdate{{Old: tip.String(), New: merged.String(), Ref: "refs/heads/" + mr.Target}},
817 })
818 return mr, nil
819}
820
821// recordPush does for a merge what Pushes.Record does for a push, as the merge
822// moved a branch: it sets pushed_at and queues the maintenance of the
823// repository. The target already moved, so a failure is logged, not returned,
824// and the caller giving up does not stop it.
825func (m *MergeRequests) recordPush(ctx context.Context, name domain.RepoName) {
826 ctx = context.WithoutCancel(ctx)
827 if _, err := touchPushedAt(ctx, m.meta, name, m.clock.Now().UTC()); err != nil {
828 log := zerowrap.FromCtx(ctx)
829 log.Error().Err(err).Str("repo", name.String()).Msg("record pushed_at after a merge")
830 }
831 m.maint.Schedule(ctx, name)
832}
833
834// moveTarget moves the target branch from tip to the head (fast-forward) or to
835// a new signed squash commit, and returns the new tip.
836func (m *MergeRequests) moveTarget(ctx context.Context, mr domain.MergeRequest, mode domain.MergeMode, tip, head domain.ObjectID) (domain.ObjectID, error) {
837 newTip := head
838 if mode == domain.MergeModeSquash {
839 commits, more, err := m.git.ListRange(ctx, mr.Repo, tip, head, domain.MaxSquashCommits)
840 if err != nil {
841 return domain.ObjectID{}, err
842 }
843 if more {
844 return domain.ObjectID{}, fmt.Errorf("%w: the branch has more than %d commits to squash", domain.ErrPreconditionFailed, domain.MaxSquashCommits)
845 }
846 tree, err := m.git.CommitTree(ctx, mr.Repo, head)
847 if err != nil {
848 return domain.ObjectID{}, err
849 }
850 commit := domain.NewSquashCommit(mr, tree, tip, commits, m.committer, m.clock.Now())
851 armored, err := m.signer.Sign(commit.Payload())
852 if err != nil {
853 return domain.ObjectID{}, fmt.Errorf("sign the squash commit of !%d in %q: %w", mr.Number, mr.Repo, err)
854 }
855 if newTip, err = m.git.WriteCommit(ctx, mr.Repo, commit.Signed(armored)); err != nil {
856 return domain.ObjectID{}, err
857 }
858 }
859 err := m.git.UpdateBranch(ctx, mr.Repo, mr.Target, tip, newTip)
860 if errors.Is(err, domain.ErrConflict) {
861 return domain.ObjectID{}, fmt.Errorf("%w: the target branch %q moved during the merge, try again", domain.ErrPreconditionFailed, mr.Target)
862 }
863 if err != nil {
864 return domain.ObjectID{}, err
865 }
866 return newTip, nil
867}
868
869// markMerged records the merge. The target already moved, so a write that
870// loses a race (a comment added meanwhile) is repeated on the fresh request.
871func (m *MergeRequests) markMerged(ctx context.Context, name domain.RepoName, number int64, mode domain.MergeMode, base, head, merged domain.ObjectID) (domain.MergeRequest, error) {
872 // The merge must be recorded even when the caller gives up.
873 ctx = context.WithoutCancel(ctx)
874 for range maxWriteRetries {
875 mr, err := m.store.Get(ctx, name, number)
876 if err != nil {
877 return domain.MergeRequest{}, fmt.Errorf("record the merge of !%d in %q: %w", number, name, err)
878 }
879 mr.State, mr.HeadID, mr.BaseID, mr.MergedCommitID, mr.MergeMode = domain.MRStateMerged, head, base, merged, mode
880 mr.UpdatedAt = m.clock.Now().UTC()
881 err = m.store.Update(ctx, mr)
882 if errors.Is(err, domain.ErrConflict) {
883 continue
884 }
885 if err != nil {
886 return domain.MergeRequest{}, fmt.Errorf("record the merge of !%d in %q: %w", number, name, err)
887 }
888 return mr, nil
889 }
890 log := zerowrap.FromCtx(ctx)
891 log.Error().Str("repo", name.String()).Int64("number", number).Str("commit", merged.String()).
892 Msg("the target branch moved but the merge request could not be marked merged")
893 return domain.MergeRequest{}, fmt.Errorf("record the merge of !%d in %q: %w", number, name, domain.ErrConflict)
894}