1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "encoding/base64"
8 "errors"
9 "fmt"
10 "math"
11 "runtime/debug"
12 "slices"
13 "strconv"
14 "strings"
15 "sync"
16
17 "github.com/bnema/zerowrap"
18
19 "bnema.dev/ermite/internal/domain"
20 "bnema.dev/ermite/internal/ports"
21)
22
23// MaxMergeRequestCommits is the most commits GetMergeRequest returns.
24const MaxMergeRequestCommits = ports.MaxMergeRequestCommits
25
26// maxWriteRetries bounds the retries of a write that loses a race with another
27// write of the same merge request. Each lost race means another writer won, so
28// a writer loses at most as many times as there are other writers at once; the
29// retry is a read and a write of ermite-kv, cheap enough for a bound this high.
30const maxWriteRetries = 32
31
32// MaxCommentsPerMergeRequest is the most comments a merge request holds. Past
33// it AddComment fails with domain.ErrResourceExhausted (resource_exhausted in
34// the API). Comments are never deleted, so the limit counts the ones ever added.
35const MaxCommentsPerMergeRequest = 1000
36
37const mrTokenPrefix = "mp1:"
38
39// MergeRequests are the merge request use cases (ADR 0017): review, comment,
40// close and merge a branch. Reading needs the access of reading the
41// repository: a repository the principal may not read, a missing one and an
42// unknown number all answer domain.ErrNotFound. Writing needs mr:write, and
43// merging mr:merge, on the repository (see scopeAccess), and that scope also
44// makes a private repository visible for the action (see writable).
45type MergeRequests struct {
46 meta ports.RepoMetaStore
47 store ports.MergeRequestStore
48 repos ports.GitRepos
49 git ports.MergeGit
50 signer ports.CommitSigner
51 committer domain.Identity
52 bus ports.EventBus
53 clock ports.Clock
54 maint ports.MaintenanceScheduler
55 ci ports.WorkflowTrigger // nil: no CI
56 checks ports.CIChecksReader // nil: no CI, so no check blocks a merge
57 pulls ports.PullRequestTrigger // nil: no pull_request workflows
58 tips ports.CIRefTipStore // nil: who merges is not recorded
59 verifier ports.CommitVerifier // nil: commits are not checked (ADR 0030)
60 mergeab MergeabilityHook // nil: no mergeability
61
62 mu sync.Mutex
63 locks map[domain.RepoName]chan struct{}
64
65 // after tracks the post-merge side effects running off the request path.
66 after sync.WaitGroup
67 // afterGate, when set by a test, runs first in the post-merge goroutine.
68 afterGate func()
69}
70
71var (
72 _ ports.MergeRequestService = (*MergeRequests)(nil)
73 _ ports.PullGate = (*MergeRequests)(nil)
74)
75
76// MergeRequestsConfig holds what the merge request use cases need. Every field
77// is required.
78type MergeRequestsConfig struct {
79 Meta ports.RepoMetaStore
80 Store ports.MergeRequestStore
81 // Repos gives the default branch of a repository.
82 Repos ports.GitRepos
83 Git ports.MergeGit
84 // Signer signs the squash commits.
85 Signer ports.CommitSigner
86 // Committer is recorded as the committer of the squash commits.
87 Committer domain.Identity
88 Bus ports.EventBus
89 Clock ports.Clock
90 // Maint queues the maintenance that follows a merge, as it does a push.
91 Maint ports.MaintenanceScheduler
92 // CI starts the workflows a merge triggers (GitHub runs "push" workflows
93 // on a merge) and Checks gives the checks the merge rule reads. Both are
94 // optional: without them there is no CI.
95 CI ports.WorkflowTrigger
96 Checks ports.CIChecksReader
97 // PullRequests starts the pull_request workflows when a request is opened,
98 // closed or merged (the push of its source branch reaches CI instead).
99 // Optional.
100 PullRequests ports.PullRequestTrigger
101 // RefTips records who moves the target branch (and deletes the record of a
102 // deleted source branch) before the merge does, as a push does
103 // (domain.CIRefTip). Optional.
104 RefTips ports.CIRefTipStore
105 // Mergeability is queued when a request is opened, closed or merged and
106 // read by Get and List. Optional: without it Mergeability stays zero.
107 Mergeability MergeabilityHook
108}
109
110// NewMergeRequests builds the merge request use cases.
111func NewMergeRequests(cfg MergeRequestsConfig) *MergeRequests {
112 return &MergeRequests{
113 meta: cfg.Meta, store: cfg.Store, repos: cfg.Repos, git: cfg.Git, signer: cfg.Signer,
114 committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock, maint: cfg.Maint, ci: cfg.CI, checks: cfg.Checks, pulls: cfg.PullRequests, tips: cfg.RefTips, mergeab: cfg.Mergeability,
115 locks: make(map[domain.RepoName]chan struct{}),
116 }
117}
118
119// WithVerifier makes GetMergeRequest fill the signature state of the commits
120// it lists (ADR 0030).
121func (m *MergeRequests) WithVerifier(v ports.CommitVerifier) *MergeRequests {
122 m.verifier = v
123 return m
124}
125
126// Wait waits for the post-merge side effects still running (source branch
127// deletion, events, CI and mergeability hooks), or for ctx to end. Call it
128// before closing what they use.
129func (m *MergeRequests) Wait(ctx context.Context) {
130 done := make(chan struct{})
131 go func() {
132 m.after.Wait()
133 close(done)
134 }()
135 select {
136 case <-done:
137 case <-ctx.Done():
138 }
139}
140
141// lockRepo takes the merge lock of a repository, waiting until ctx ends. The
142// merges of one repository, and the creation of its merge requests, run one at
143// a time. The returned function releases the lock.
144func (m *MergeRequests) lockRepo(ctx context.Context, name domain.RepoName) (unlock func(), err error) {
145 m.mu.Lock()
146 ch, ok := m.locks[name]
147 if !ok {
148 ch = make(chan struct{}, 1)
149 m.locks[name] = ch
150 }
151 m.mu.Unlock()
152 select {
153 case ch <- struct{}{}:
154 return func() { <-ch }, nil
155 case <-ctx.Done():
156 return nil, ctx.Err()
157 }
158}
159
160// readable looks the repository up and applies the read rule: reading merge
161// requests needs the access of reading the repository (canRead). repo:write
162// alone does not read them, as it does not fetch a private repository either.
163func (m *MergeRequests) readable(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
164 repo, err := lookupRepo(ctx, m.meta, name)
165 if err != nil {
166 return domain.Repo{}, err
167 }
168 if !canRead(p, repo) {
169 return domain.Repo{}, domain.ErrNotFound
170 }
171 return repo, nil
172}
173
174// writable looks the repository up for an action that needs a merge request
175// scope. A principal whose token holds that scope on the repository sees the
176// repository for this action even without repo:read, so that the agent preset
177// (repo:write, mr:write) can open a request on the private repository it just
178// pushed. A principal with neither repo:read nor the scope, and an unknown
179// repository, still get domain.ErrNotFound; one that can read the repository
180// but lacks the scope gets the error of scopeAccess.
181func (m *MergeRequests) writable(ctx context.Context, p domain.Principal, name domain.RepoName, scope domain.Scope) (domain.Repo, error) {
182 repo, err := lookupRepo(ctx, m.meta, name)
183 if err != nil {
184 return domain.Repo{}, err
185 }
186 if !canRead(p, repo) && !p.Allows(scope, name) {
187 return domain.Repo{}, domain.ErrNotFound
188 }
189 if err := scopeAccess(p, repo, scope); err != nil {
190 return domain.Repo{}, err
191 }
192 return repo, nil
193}
194
195// withHead fills HeadID of an open request from its source branch: zero when
196// the branch is gone. Merged and closed requests keep the head they stored.
197func (m *MergeRequests) withHead(ctx context.Context, mr domain.MergeRequest) (domain.MergeRequest, error) {
198 if !mr.IsOpen() {
199 return mr, nil
200 }
201 head, err := m.git.BranchHead(ctx, mr.Repo, mr.Source)
202 switch {
203 case errors.Is(err, domain.ErrNotFound):
204 mr.HeadID = domain.ObjectID{}
205 case err != nil:
206 return domain.MergeRequest{}, err
207 default:
208 mr.HeadID = head
209 }
210 return mr, nil
211}
212
213// branchOrZero returns the head of a branch, the zero id when it is gone.
214func (m *MergeRequests) branchOrZero(ctx context.Context, repo domain.RepoName, branch string) (domain.ObjectID, error) {
215 id, err := m.git.BranchHead(ctx, repo, branch)
216 if errors.Is(err, domain.ErrNotFound) {
217 return domain.ObjectID{}, nil
218 }
219 return id, err
220}
221
222// Create opens a merge request and publishes MergeRequestOpened. The principal
223// needs mr:write. A request that breaks a rule of ADR 0017 (the texts, a
224// branch that does not exist, source and target the same) is rejected whole
225// with an *domain.InvalidFieldsError naming the fields. An open request with
226// the same source and target gives domain.ErrAlreadyExists.
227func (m *MergeRequests) Create(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.CreateMergeRequestParams) (domain.MergeRequest, error) {
228 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
229 if err != nil {
230 return domain.MergeRequest{}, err
231 }
232 var v domain.Violations
233 title, err := domain.ParseMRTitle(params.Title)
234 if err != nil {
235 v.AddErr("title", err)
236 }
237 desc, err := domain.ParsePlainText(params.Description)
238 if err != nil {
239 v.AddErr("description", err)
240 }
241 // Under the lock, so that two requests for the same pair cannot both pass
242 // the check for an existing one, and a merge that deletes its source
243 // cannot remove a branch between its check here and the creation.
244 unlock, err := m.lockRepo(ctx, name)
245 if err != nil {
246 return domain.MergeRequest{}, err
247 }
248 defer unlock()
249 source, target, err := m.checkBranches(ctx, name, params, &v)
250 if err != nil {
251 return domain.MergeRequest{}, err
252 }
253 if err := v.Err(); err != nil {
254 return domain.MergeRequest{}, err
255 }
256 existing, err := m.store.List(ctx, name)
257 if err != nil {
258 return domain.MergeRequest{}, fmt.Errorf("list merge requests of %q: %w", name, err)
259 }
260 for _, e := range existing {
261 if e.IsOpen() && e.Source == source && e.Target == target {
262 return domain.MergeRequest{}, fmt.Errorf("%w: !%d already merges %s into %s", domain.ErrAlreadyExists, e.Number, source, target)
263 }
264 }
265
266 now := m.clock.Now().UTC()
267 mr, err := m.store.Create(ctx, domain.MergeRequest{
268 Repo: name, Title: title, Description: desc, Source: source, Target: target,
269 State: domain.MRStateOpen, Author: domain.AuthorName(p), CreatedAt: now, UpdatedAt: now,
270 })
271 if err != nil {
272 return domain.MergeRequest{}, fmt.Errorf("create merge request of %q: %w", name, err)
273 }
274 log := zerowrap.FromCtx(ctx)
275 log.Info().Str("repo", name.String()).Int64("number", mr.Number).
276 Str("source", source).Str("target", target).
277 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request opened")
278 mr, err = m.withHead(ctx, mr)
279 if err != nil {
280 return domain.MergeRequest{}, err
281 }
282 recordActivity(ctx, m.meta, repo.Name, now)
283 m.bus.Publish(ctx, domain.MergeRequestOpened{MergeRequest: mr, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
284 if m.mergeab != nil {
285 m.mergeab.OnOpened(ctx, mr)
286 }
287 if m.pulls != nil {
288 // The commits the opening is for are the ones of now.
289 base, _ := m.branchOrZero(ctx, name, target)
290 m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: mr.Number, Action: domain.PullOpened, Head: mr.HeadID, Base: base}, p)
291 }
292 return mr, nil
293}
294
295// checkBranches validates the branches of a new request and returns them, with
296// the default branch filled in for the target.
297//
298// Only a branch that does not exist is a violation. Any other failure to read
299// Git is returned as an error, so that it is logged and masked, not shown in
300// the field violations.
301func (m *MergeRequests) checkBranches(ctx context.Context, name domain.RepoName, params ports.CreateMergeRequestParams, v *domain.Violations) (source, target string, err error) {
302 source, target = params.Source, params.Target
303 if source == "" {
304 v.Add("source_branch", "is required")
305 } else if _, err := domain.ParseBranchName(source); err != nil {
306 v.Add("source_branch", "is not a valid branch name")
307 source = ""
308 }
309 if target == "" {
310 def, exists, err := m.repos.DefaultBranch(ctx, name)
311 switch {
312 case err != nil: // includes domain.ErrNotFound: the repository is gone
313 return "", "", fmt.Errorf("read the default branch of %q: %w", name, err)
314 case !exists:
315 v.Add("target_branch", fmt.Sprintf("is empty and the default branch %q does not exist", def))
316 default:
317 target = def
318 }
319 } else if _, err := domain.ParseBranchName(target); err != nil {
320 v.Add("target_branch", "is not a valid branch name")
321 target = ""
322 }
323 if source != "" && source == target {
324 v.Add("target_branch", "must differ from source_branch")
325 return source, target, nil
326 }
327 for _, b := range []struct{ field, name string }{{"source_branch", source}, {"target_branch", target}} {
328 if b.name == "" {
329 continue
330 }
331 _, err := m.git.BranchHead(ctx, name, b.name)
332 if errors.Is(err, domain.ErrNotFound) {
333 v.Add(b.field, fmt.Sprintf("branch %q does not exist", b.name))
334 } else if err != nil {
335 return "", "", fmt.Errorf("read branch %q of %q: %w", b.name, name, err)
336 }
337 }
338 return source, target, nil
339}
340
341// List returns a page of the merge requests of a repository, highest number
342// first. The open ones carry the head of their source branch. The page token
343// names the last number of the previous page, so requests created between two
344// pages never shift the others.
345func (m *MergeRequests) List(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.ListMergeRequestsParams) (ports.MergeRequestPage, error) {
346 size, err := pageSize(params.PageSize)
347 if err != nil {
348 return ports.MergeRequestPage{}, err
349 }
350 after, err := decodeMRToken(params.PageToken)
351 if err != nil {
352 return ports.MergeRequestPage{}, err
353 }
354 if _, err := m.readable(ctx, p, name); err != nil {
355 return ports.MergeRequestPage{}, err
356 }
357 all, err := m.store.List(ctx, name)
358 if err != nil {
359 return ports.MergeRequestPage{}, fmt.Errorf("list merge requests of %q: %w", name, err)
360 }
361 var page ports.MergeRequestPage
362 page.MergeRequests, page.NextPageToken = pageAfter(all, size,
363 func(mr domain.MergeRequest) bool {
364 return (params.State == "" || mr.State == params.State) && (after == 0 || mr.Number < after)
365 },
366 encodeMRToken)
367 tips := map[string]domain.ObjectID{} // one BranchHead per distinct target
368 for i, mr := range page.MergeRequests {
369 if page.MergeRequests[i], err = m.withHead(ctx, mr); err != nil {
370 return ports.MergeRequestPage{}, err
371 }
372 if m.mergeab == nil || !mr.IsOpen() {
373 continue
374 }
375 tip, ok := tips[mr.Target]
376 if !ok {
377 if tip, err = m.branchOrZero(ctx, name, mr.Target); err != nil {
378 return ports.MergeRequestPage{}, err
379 }
380 tips[mr.Target] = tip
381 }
382 if page.Mergeability == nil {
383 page.Mergeability = map[int64]domain.Mergeability{}
384 }
385 page.Mergeability[mr.Number] = m.mergeab.Read(ctx, page.MergeRequests[i], tip)
386 }
387 return page, nil
388}
389
390// MarkReady implements ports.MergeRequestService: the title of an open draft
391// loses its draft prefix, so the request can be merged. The principal needs
392// mr:write. A request that is not open, or not a draft, gives
393// domain.ErrPreconditionFailed.
394func (m *MergeRequests) MarkReady(ctx context.Context, p domain.Principal, name domain.RepoName, number int64) (domain.MergeRequest, error) {
395 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
396 if err != nil {
397 return domain.MergeRequest{}, err
398 }
399 unlock, err := m.lockRepo(ctx, name) // a merge sees the request as it is after
400 if err != nil {
401 return domain.MergeRequest{}, err
402 }
403 defer unlock()
404 for range maxWriteRetries {
405 mr, err := m.store.Get(ctx, name, number)
406 if err != nil {
407 return domain.MergeRequest{}, err
408 }
409 if !mr.IsOpen() {
410 return domain.MergeRequest{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
411 }
412 title, ok := domain.ReadyTitle(mr.Title)
413 if !ok {
414 return domain.MergeRequest{}, fmt.Errorf("%w: !%d is not a draft", domain.ErrPreconditionFailed, number)
415 }
416 mr.Title, mr.UpdatedAt = title, m.clock.Now().UTC()
417 err = m.store.Update(ctx, mr)
418 if errors.Is(err, domain.ErrConflict) {
419 continue
420 }
421 if err != nil {
422 return domain.MergeRequest{}, fmt.Errorf("mark !%d of %q ready: %w", number, name, err)
423 }
424 recordActivity(ctx, m.meta, repo.Name, mr.UpdatedAt)
425 m.bus.Publish(ctx, domain.MergeRequestReady{MergeRequest: mr, At: mr.UpdatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
426 return m.withHead(ctx, mr)
427 }
428 return domain.MergeRequest{}, fmt.Errorf("mark !%d of %q ready: %w", number, name, domain.ErrConflict)
429}
430
431// CountOpen implements ports.MergeRequestService.
432func (m *MergeRequests) CountOpen(ctx context.Context, p domain.Principal, name domain.RepoName) (int, error) {
433 if _, err := m.readable(ctx, p, name); err != nil {
434 return 0, err
435 }
436 n, err := m.store.CountOpen(ctx, name)
437 if err != nil {
438 return 0, fmt.Errorf("count open merge requests of %q: %w", name, err)
439 }
440 return n, nil
441}
442
443// Mergeability implements ports.MergeRequestService: the mergeability of an
444// open merge request for the current tips, queued when there is no up-to-date
445// result. It is the zero value for a request that is not open, or without a
446// mergeability service.
447func (m *MergeRequests) Mergeability(ctx context.Context, p domain.Principal, name domain.RepoName, number int64) (domain.Mergeability, error) {
448 _, mr, err := m.get(ctx, p, name, number)
449 if err != nil {
450 return domain.Mergeability{}, err
451 }
452 if m.mergeab == nil || !mr.IsOpen() {
453 return domain.Mergeability{}, nil
454 }
455 if mr, err = m.withHead(ctx, mr); err != nil {
456 return domain.Mergeability{}, err
457 }
458 tip, err := m.branchOrZero(ctx, name, mr.Target)
459 if err != nil {
460 return domain.Mergeability{}, err
461 }
462 return m.mergeab.Read(ctx, mr, tip), nil
463}
464
465func encodeMRToken(last domain.MergeRequest) string {
466 return base64.RawURLEncoding.EncodeToString([]byte(mrTokenPrefix + strconv.FormatInt(last.Number, 10)))
467}
468
469// decodeMRToken returns 0 for the empty token.
470func decodeMRToken(tok string) (int64, error) {
471 if tok == "" {
472 return 0, nil
473 }
474 raw, ok := decodeToken(tok, mrTokenPrefix)
475 if ok {
476 if n, err := strconv.ParseInt(raw, 10, 64); err == nil && n > 0 {
477 return n, nil
478 }
479 }
480 return 0, fmt.Errorf("%w: merge requests", domain.ErrInvalidPageToken)
481}
482
483// get reads a merge request of a repository the principal can read.
484func (m *MergeRequests) get(ctx context.Context, p domain.Principal, name domain.RepoName, number int64) (domain.Repo, domain.MergeRequest, error) {
485 repo, err := m.readable(ctx, p, name)
486 if err != nil {
487 return domain.Repo{}, domain.MergeRequest{}, err
488 }
489 mr, err := m.store.Get(ctx, name, number)
490 if err != nil {
491 return domain.Repo{}, domain.MergeRequest{}, err
492 }
493 return repo, mr, nil
494}
495
496// windowLimits is the diff limits of a read: nil gives
497// domain.DefaultDiffLimits (the API contract, generated files read); a window
498// lists every file (MaxFiles 0), reads From..To and skips generated files; a
499// Full window reads the one file From with domain.OneFileDiffLimits. An invalid
500// window (negative, To < From, Full without To == From+1) is domain.ErrInvalid.
501func windowLimits(w *ports.FileWindow) (domain.DiffLimits, error) {
502 if w == nil {
503 return domain.DefaultDiffLimits(), nil
504 }
505 if w.From < 0 || w.To < 0 || (w.To != 0 && w.To < w.From) || (w.Full && w.To != w.From+1) {
506 return domain.DiffLimits{}, fmt.Errorf("%w: file window [%d, %d) full=%t", domain.ErrInvalid, w.From, w.To, w.Full)
507 }
508 if w.Full {
509 return domain.OneFileDiffLimits(w.From), nil
510 }
511 l := domain.DefaultDiffLimits()
512 l.MaxFiles, l.From, l.To, l.SkipGenerated = 0, w.From, w.To, true
513 return l, nil
514}
515
516// Get returns a merge request with its commits, its diff, its comments and the
517// read position. The diff compares the merge base of the target and the head
518// with the head ("target...head"), within domain.DefaultDiffLimits. With
519// opts.SinceLastRead it compares the head recorded by MarkRead with the
520// current head instead, which works after a force-push as long as the recorded
521// commit still exists; when none was recorded the diff is the default one, and
522// when the recorded commit is gone too, with LastReadLost set. A merged or
523// closed request keeps the head and the base it had.
524//
525// opts.Files lists every changed file and reads the hunks of a window of them
526// only; opts.Base and opts.Head make Get fail with domain.ErrConflict when the
527// diff is no longer the one the caller shows (before any diff is read);
528// opts.DiffOnly skips the commits. An invalid window gives domain.ErrInvalid.
529func (m *MergeRequests) Get(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, opts ports.GetMergeRequestOptions) (ports.MergeRequestDetail, error) {
530 limits, err := windowLimits(opts.Files)
531 if err != nil {
532 return ports.MergeRequestDetail{}, err
533 }
534 _, mr, err := m.get(ctx, p, name, number)
535 if err != nil {
536 return ports.MergeRequestDetail{}, err
537 }
538 if mr, err = m.withHead(ctx, mr); err != nil {
539 return ports.MergeRequestDetail{}, err
540 }
541 if !opts.Head.IsZero() && opts.Head != mr.HeadID {
542 return ports.MergeRequestDetail{}, fmt.Errorf("%w: the head of !%d moved to %s", domain.ErrConflict, number, mr.HeadID)
543 }
544 // checkBase fails when the caller shows another diff base than base.
545 checkBase := func(base domain.ObjectID) error {
546 if !opts.Base.IsZero() && opts.Base != base {
547 return fmt.Errorf("%w: the diff base of !%d is now %s", domain.ErrConflict, number, base)
548 }
549 return nil
550 }
551 d := ports.MergeRequestDetail{MergeRequest: mr}
552 if m.mergeab != nil && mr.IsOpen() {
553 tip, err := m.branchOrZero(ctx, name, mr.Target)
554 if err != nil {
555 return ports.MergeRequestDetail{}, err
556 }
557 d.Mergeability = m.mergeab.Read(ctx, mr, tip)
558 }
559 if d.Comments, err = m.store.ListComments(ctx, name, number); err != nil {
560 return ports.MergeRequestDetail{}, fmt.Errorf("list comments of !%d in %q: %w", number, name, err)
561 }
562 d.Verdicts = domain.CurrentVerdicts(d.Comments, mr.HeadID)
563 if d.LastReadID, err = m.store.GetRead(ctx, name, number); err != nil {
564 return ports.MergeRequestDetail{}, err
565 }
566 d.HeadUnread = !mr.HeadID.IsZero() && mr.HeadID != d.LastReadID
567
568 exclude, base, err := m.rangeOf(ctx, mr)
569 if err != nil {
570 return ports.MergeRequestDetail{}, err
571 }
572 head := mr.HeadID
573 if head.IsZero() {
574 return d, checkBase(domain.ObjectID{}) // the source branch is gone: no commits, no diff
575 }
576 if !mr.IsOpen() {
577 // After a squash the reviewed commits are only reachable from the
578 // source branch; once it is deleted, gc may prune them.
579 switch ok, err := m.git.HasCommit(ctx, name, head); {
580 case err != nil:
581 return ports.MergeRequestDetail{}, err
582 case !ok:
583 // The reviewed commits are gone: no commits, no diff.
584 if err := checkBase(domain.ObjectID{}); err != nil {
585 return ports.MergeRequestDetail{}, err
586 }
587 return d, nil
588 }
589 }
590 if !opts.DiffOnly {
591 if d.Commits, d.CommitsTruncated, err = m.git.ListRange(ctx, name, exclude, head, MaxMergeRequestCommits); err != nil {
592 return ports.MergeRequestDetail{}, err
593 }
594 verifyCommits(ctx, m.verifier, name, d.Commits)
595 if !d.LastReadID.IsZero() {
596 for i, c := range d.Commits {
597 if c.ID == d.LastReadID {
598 d.NewCommits = i
599 break
600 }
601 }
602 }
603 }
604 if opts.SinceLastRead && !d.LastReadID.IsZero() {
605 switch ok, err := m.git.HasCommit(ctx, name, d.LastReadID); {
606 case err != nil:
607 return ports.MergeRequestDetail{}, err
608 case ok:
609 base = d.LastReadID
610 default:
611 d.LastReadLost = true
612 }
613 }
614 if err := checkBase(base); err != nil {
615 return ports.MergeRequestDetail{}, err
616 }
617 if base.IsZero() {
618 return d, nil
619 }
620 d.DiffBaseID = base
621 if d.Stats, d.Diff, err = m.git.DiffCommits(ctx, name, base, head, limits); err != nil {
622 return ports.MergeRequestDetail{}, err
623 }
624 // The page of a client without scripts reads one collapsed file in full.
625 if i := opts.Load; i != nil && *i >= 0 && *i < len(d.Diff.Files) && d.Diff.Files[*i].Omitted.Collapsed() {
626 _, one, err := m.git.DiffCommits(ctx, name, base, head, domain.OneFileDiffLimits(*i))
627 if err != nil {
628 return ports.MergeRequestDetail{}, err
629 }
630 if *i < len(one.Files) {
631 d.Diff.Files[*i] = one.Files[*i]
632 }
633 }
634 return d, nil
635}
636
637// rangeOf returns what the review of a request compares: the commit whose
638// history the request's commits exclude, and the base of its diff. For an open
639// request they are the tip of the target and the merge base of the target and
640// the head; both are zero when the target is gone or shares no history with
641// the head. A merged or closed request keeps the base it stored for both.
642func (m *MergeRequests) rangeOf(ctx context.Context, mr domain.MergeRequest) (exclude, base domain.ObjectID, err error) {
643 if !mr.IsOpen() {
644 return mr.BaseID, mr.BaseID, nil
645 }
646 if mr.HeadID.IsZero() {
647 return domain.ObjectID{}, domain.ObjectID{}, nil
648 }
649 tip, err := m.branchOrZero(ctx, mr.Repo, mr.Target)
650 if err != nil || tip.IsZero() {
651 return domain.ObjectID{}, domain.ObjectID{}, err
652 }
653 base, err = m.git.MergeBase(ctx, mr.Repo, tip, mr.HeadID)
654 if errors.Is(err, domain.ErrNotFound) {
655 return tip, domain.ObjectID{}, nil
656 }
657 return tip, base, err
658}
659
660// AddComment adds a comment and publishes CommentAdded. The principal needs
661// mr:write. The text must follow the rules of ADR 0032 and not be blank unless carrying a verdict; a
662// comment on a line must name a line of the current diff of the request (see
663// domain.Diff.HasLine), else the call fails with domain.ErrInvalid. A request
664// that is not open gives domain.ErrPreconditionFailed, and one that already
665// holds MaxCommentsPerMergeRequest comments domain.ErrResourceExhausted.
666func (m *MergeRequests) AddComment(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, params ports.AddCommentParams) (domain.Comment, error) {
667 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
668 if err != nil {
669 return domain.Comment{}, err
670 }
671 mr, err := m.store.Get(ctx, name, number)
672 if err != nil {
673 return domain.Comment{}, err
674 }
675 var v domain.Violations
676 text, err := domain.ParsePlainText(params.Text)
677 switch {
678 case err != nil:
679 v.AddErr("text", err)
680 case text.IsBlank() && params.Verdict == domain.VerdictNone:
681 v.Add("text", "is required")
682 }
683 general := params.Path == ""
684 switch {
685 case general && (params.Line != 0 || params.Side != domain.SideNone || params.StartLine != 0):
686 v.Add("path", "is required with line and side")
687 case !general && params.Line < 1:
688 v.Add("line", "must be 1 or more with a path")
689 case !general && params.Side != domain.SideOld && params.Side != domain.SideNew:
690 v.Add("side", "is required with a path")
691 }
692 if params.StartLine != 0 && (params.StartLine < 1 || params.StartLine >= params.Line) {
693 v.Add("start_line", "must be 1 or more and less than line")
694 }
695 if params.Verdict > domain.VerdictChangesRequested {
696 v.Add("verdict", "is unknown")
697 }
698 if params.ReplyTo != 0 {
699 if params.Path != "" || params.Line != 0 || params.StartLine != 0 || params.Side != domain.SideNone || params.Verdict != domain.VerdictNone {
700 v.Add("reply_to", "a reply cannot specify an anchor or verdict")
701 }
702 }
703 if err := v.Err(); err != nil {
704 return domain.Comment{}, err
705 }
706 if params.ReplyTo != 0 {
707 root, err := m.replyRoot(ctx, name, number, params.ReplyTo)
708 if err != nil {
709 return domain.Comment{}, err
710 }
711 params.Path, params.Line, params.StartLine, params.Side = root.Path, root.Line, root.StartLine, root.Side
712 }
713 if params.Verdict != domain.VerdictNone {
714 if err := scopeAccess(p, repo, domain.ScopeMRReview); err != nil {
715 return domain.Comment{}, err
716 }
717 }
718 if !mr.IsOpen() {
719 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
720 }
721
722 if mr, err = m.withHead(ctx, mr); err != nil {
723 return domain.Comment{}, err
724 }
725 if !general && params.ReplyTo == 0 {
726 if err := m.checkLine(ctx, mr, params); err != nil {
727 return domain.Comment{}, err
728 }
729 }
730
731 var c domain.Comment
732 for range maxWriteRetries {
733 if mr.CommentSeq >= MaxCommentsPerMergeRequest {
734 return domain.Comment{}, fmt.Errorf("%w: !%d already has %d comments", domain.ErrResourceExhausted, number, MaxCommentsPerMergeRequest)
735 }
736 c = domain.Comment{
737 ID: mr.CommentSeq + 1, Author: domain.AuthorName(p), Path: params.Path, Line: params.Line, Side: params.Side,
738 ReplyTo: params.ReplyTo, StartLine: params.StartLine, Verdict: params.Verdict, Resolved: params.Verdict == domain.VerdictApprove,
739 CommitID: mr.HeadID, Text: text, CreatedAt: m.clock.Now().UTC(),
740 }
741 mr.CommentSeq, mr.UpdatedAt = c.ID, c.CreatedAt
742 err = m.store.SaveComment(ctx, mr, c)
743 if !errors.Is(err, domain.ErrConflict) {
744 break
745 }
746 // Another write got in between: read the request again.
747 fresh, gerr := m.store.Get(ctx, name, number)
748 if gerr != nil {
749 return domain.Comment{}, gerr
750 }
751 if !fresh.IsOpen() {
752 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, fresh.State)
753 }
754 if params.ReplyTo != 0 {
755 // The root may have been deleted by the write that got in between.
756 if _, rerr := m.replyRoot(ctx, name, number, params.ReplyTo); rerr != nil {
757 return domain.Comment{}, rerr
758 }
759 }
760 fresh.HeadID = mr.HeadID
761 mr = fresh
762 }
763 if err != nil {
764 return domain.Comment{}, fmt.Errorf("save comment on !%d of %q: %w", number, name, err)
765 }
766 log := zerowrap.FromCtx(ctx)
767 log.Info().Str("repo", name.String()).Int64("number", number).Int64("comment", c.ID).
768 Int64("reply_to", c.ReplyTo).Str("verdict", c.Verdict.String()).Bool("on_line", !general).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("comment added")
769 recordActivity(ctx, m.meta, name, c.CreatedAt)
770 m.bus.Publish(ctx, domain.CommentAdded{
771 Repo: name, Number: number, Comment: c, At: c.CreatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
772 })
773 return c, nil
774}
775
776// checkLine verifies that a comment designates a line of the current diff of
777// the request.
778func (m *MergeRequests) checkLine(ctx context.Context, mr domain.MergeRequest, params ports.AddCommentParams) error {
779 var v domain.Violations
780 _, base, err := m.rangeOf(ctx, mr)
781 if err != nil {
782 return err
783 }
784 if mr.HeadID.IsZero() || base.IsZero() {
785 v.Add("path", "the merge request has no diff to comment on")
786 return v.Err()
787 }
788 // A comment may be on a file past the 300th that the web view lists, so
789 // the file is looked up among all the files (MaxFiles 0). Reading every
790 // file's hunks within the default total limits would reject a comment on a
791 // file the earlier ones left no budget for. So a first read lists the files
792 // with no hunks (From past the end: cheap, git is stopped at once), and a
793 // second loads the one file in full (a Full window: the byte guard and
794 // domain.MaxOneFileLines only, generated files included), as the web view
795 // does when it loads a collapsed file: any line the page can show can be
796 // commented on, and a file over those limits takes no line comment.
797 list := domain.DefaultDiffLimits()
798 list.MaxFiles, list.From = 0, math.MaxInt
799 _, diff, err := m.git.DiffCommits(ctx, mr.Repo, base, mr.HeadID, list)
800 if err != nil {
801 return err
802 }
803 idx := slices.IndexFunc(diff.Files, func(f domain.FileDiff) bool { return f.CommentPath() == params.Path })
804 if idx >= 0 {
805 one, err := windowLimits(&ports.FileWindow{From: idx, To: idx + 1, Full: true})
806 if err != nil {
807 return err
808 }
809 if _, diff, err = m.git.DiffCommits(ctx, mr.Repo, base, mr.HeadID, one); err != nil {
810 return err
811 }
812 }
813 if idx < 0 || !diff.HasLine(params.Path, params.Side, params.Line) {
814 v.Add("path", fmt.Sprintf("%s:%d on the %s side is not a line of the diff", params.Path, params.Line, params.Side))
815 }
816 if params.StartLine != 0 && !diff.HasLine(params.Path, params.Side, params.StartLine) {
817 v.Add("start_line", "is not a line of the diff")
818 }
819 return v.Err()
820}
821
822// ResolveComment marks a comment resolved or unresolved. The principal needs
823// mr:write. An unknown comment gives domain.ErrNotFound and a request that is
824// not open domain.ErrPreconditionFailed. Setting the state the comment already
825// has succeeds and writes nothing.
826func (m *MergeRequests) ResolveComment(ctx context.Context, p domain.Principal, name domain.RepoName, number, commentID int64, resolved bool) (domain.Comment, error) {
827 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
828 if err != nil {
829 return domain.Comment{}, err
830 }
831 for range maxWriteRetries {
832 mr, err := m.store.Get(ctx, name, number)
833 if err != nil {
834 return domain.Comment{}, err
835 }
836 c, err := m.store.GetComment(ctx, name, number, commentID)
837 if err != nil {
838 return domain.Comment{}, err
839 }
840 if !mr.IsOpen() {
841 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
842 }
843 if c.IsReply() {
844 var v domain.Violations
845 v.Add("comment_id", fmt.Sprintf("resolve the thread on its root ^%d", c.ReplyTo))
846 return domain.Comment{}, v.Err()
847 }
848 if c.Resolved == resolved {
849 return c, nil
850 }
851 c.Resolved = resolved
852 mr.UpdatedAt = m.clock.Now().UTC()
853 err = m.store.SaveComment(ctx, mr, c)
854 if errors.Is(err, domain.ErrConflict) {
855 continue
856 }
857 if err != nil {
858 return domain.Comment{}, fmt.Errorf("resolve comment %d of !%d in %q: %w", commentID, number, name, err)
859 }
860 m.bus.Publish(ctx, domain.CommentResolved{Repo: name, Number: number, Comment: c, At: mr.UpdatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
861 return c, nil
862 }
863 return domain.Comment{}, fmt.Errorf("resolve comment %d of !%d in %q: %w", commentID, number, name, domain.ErrConflict)
864}
865
866// Close closes an open merge request without merging it and publishes
867// MergeRequestClosed. The principal needs mr:write. A request that is not open
868// gives domain.ErrPreconditionFailed. The request keeps the head it had and
869// its merge base, so that its diff stays readable.
870func (m *MergeRequests) Close(ctx context.Context, p domain.Principal, name domain.RepoName, number int64) (domain.MergeRequest, error) {
871 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
872 if err != nil {
873 return domain.MergeRequest{}, err
874 }
875 unlock, err := m.lockRepo(ctx, name) // a merge and a close of one request exclude each other
876 if err != nil {
877 return domain.MergeRequest{}, err
878 }
879 defer unlock()
880 for range maxWriteRetries {
881 mr, err := m.store.Get(ctx, name, number)
882 if err != nil {
883 return domain.MergeRequest{}, err
884 }
885 if !mr.IsOpen() {
886 return domain.MergeRequest{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
887 }
888 if mr, err = m.withHead(ctx, mr); err != nil {
889 return domain.MergeRequest{}, err
890 }
891 _, base, err := m.rangeOf(ctx, mr)
892 if err != nil {
893 return domain.MergeRequest{}, err
894 }
895 mr.State, mr.BaseID, mr.UpdatedAt = domain.MRStateClosed, base, m.clock.Now().UTC()
896 err = m.store.Update(ctx, mr)
897 if errors.Is(err, domain.ErrConflict) {
898 continue
899 }
900 if err != nil {
901 return domain.MergeRequest{}, fmt.Errorf("close !%d of %q: %w", number, name, err)
902 }
903 log := zerowrap.FromCtx(ctx)
904 log.Info().Str("repo", name.String()).Int64("number", number).
905 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request closed")
906 recordActivity(ctx, m.meta, repo.Name, mr.UpdatedAt)
907 m.bus.Publish(ctx, domain.MergeRequestClosed{MergeRequest: mr, At: mr.UpdatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
908 if m.pulls != nil {
909 m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID}, p)
910 }
911 return mr, nil
912 }
913 return domain.MergeRequest{}, fmt.Errorf("close !%d of %q: %w", number, name, domain.ErrConflict)
914}
915
916// MarkRead records the head the reviewer has read, for the since-last-read diff
917// of Get. The principal needs mr:write. A zero head means the current head of
918// the source branch (the stored one for a merged or closed request); one that
919// is not a commit of the repository gives domain.ErrInvalid. It returns the
920// head recorded.
921func (m *MergeRequests) MarkRead(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, head domain.ObjectID) (domain.ObjectID, error) {
922 if _, err := m.writable(ctx, p, name, domain.ScopeMRWrite); err != nil {
923 return domain.ObjectID{}, err
924 }
925 mr, err := m.store.Get(ctx, name, number)
926 if err != nil {
927 return domain.ObjectID{}, err
928 }
929 if head.IsZero() {
930 if mr, err = m.withHead(ctx, mr); err != nil {
931 return domain.ObjectID{}, err
932 }
933 if head = mr.HeadID; head.IsZero() {
934 return domain.ObjectID{}, fmt.Errorf("%w: the source branch %q no longer exists, give a head", domain.ErrPreconditionFailed, mr.Source)
935 }
936 } else {
937 var v domain.Violations
938 switch ok, err := m.git.HasCommit(ctx, name, head); {
939 case err != nil:
940 return domain.ObjectID{}, err
941 case !ok:
942 v.Add("head_id", fmt.Sprintf("%s is not a commit of the repository", head))
943 return domain.ObjectID{}, v.Err()
944 }
945 }
946 if err := m.store.SetRead(ctx, name, number, head); err != nil {
947 return domain.ObjectID{}, err
948 }
949 return head, nil
950}
951
952// Merge merges an open merge request, publishes MergeRequestMerged then a
953// PushEvent for the target branch, and returns the merged request. The
954// principal needs mr:merge. Merges of one repository run one at a time.
955//
956// It fails with domain.ErrInvalid for a zero head, and with
957// domain.ErrPreconditionFailed when the request is not open, the mode is not
958// allowed by the settings of the repository, the source branch no longer
959// points to the head ("branch moved, read again") or is gone, the target is
960// gone, the mode's own condition does not hold (see below), or the branch
961// brings nothing. Checks come with CI (CI-1): there are none yet, so the
962// setting that lets failing checks block the merge has nothing to act on.
963//
964// The conditions are always checked with Git, under the lock, never from the
965// stored mergeability. A fast-forward needs the target to be an ancestor of
966// the head (the author rebases and pushes again) and moves the target to the
967// head, with the old value as a precondition. A squash and a merge commit need
968// the head to merge cleanly into the target (a behind branch is fine). A squash
969// writes one commit with the merged tree, the target as parent and the author
970// and message of domain.NewSquashCommit; a merge commit has two parents (the
971// target, then the head) and the message of domain.NewMergeCommit. Both are
972// signed by the instance key and move the target the same way.
973//
974// With DeleteSource the merge is also refused, before anything changes, when
975// the source is the default branch or the source or target of another open
976// request.
977//
978// Only the merge itself runs before Merge returns, under the lock. Afterwards,
979// off the request path and without the lock: the deletion of the source branch
980// (if it still points to the head; a failure is logged and leaves the branch),
981// MergeRequestMerged then a PushEvent (which carries the deletion too), and
982// the CI and mergeability hooks. Wait waits for them. SourceDeleted therefore
983// only says that the deletion was requested.
984func (m *MergeRequests) Merge(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, params ports.MergeParams) (ports.MergeResult, error) {
985 mr, err := m.merge(ctx, p, name, number, params)
986 return ports.MergeResult{MergeRequest: mr, SourceDeleted: err == nil && params.DeleteSource}, err
987}
988
989// maxConflictPathsShown is how many conflicting paths a refusal lists.
990const maxConflictPathsShown = 5
991
992// checkMode is the authoritative per-mode check of a merge, done with Git. It
993// returns the tree a squash or merge commit gets (zero for a fast-forward).
994func (m *MergeRequests) checkMode(ctx context.Context, mr domain.MergeRequest, mode domain.MergeMode, tip, head domain.ObjectID) (domain.ObjectID, error) {
995 if mode == domain.MergeModeFastForward {
996 ok, err := m.git.IsAncestor(ctx, mr.Repo, tip, head)
997 if err != nil {
998 return domain.ObjectID{}, err
999 }
1000 if !ok {
1001 return domain.ObjectID{}, fmt.Errorf("%w: the target has moved: rebase and push again, or merge with squash or a merge commit", domain.ErrPreconditionFailed)
1002 }
1003 return domain.ObjectID{}, nil
1004 }
1005 res, err := m.git.MergeTree(ctx, mr.Repo, tip, head)
1006 if err != nil {
1007 return domain.ObjectID{}, err
1008 }
1009 res = res.Bounded()
1010 if !res.Clean {
1011 what := "no common history"
1012 if len(res.ConflictPaths) > 0 {
1013 shown := res.ConflictPaths
1014 if len(shown) > maxConflictPathsShown {
1015 shown = shown[:maxConflictPathsShown]
1016 }
1017 what = strings.Join(shown, ", ")
1018 if len(shown) < len(res.ConflictPaths) || res.Truncated {
1019 what += ", ..."
1020 }
1021 }
1022 return domain.ObjectID{}, fmt.Errorf("%w: conflicts with %s: %s: rebase and push again", domain.ErrPreconditionFailed, mr.Target, what)
1023 }
1024 return res.Tree, nil
1025}
1026
1027func (m *MergeRequests) merge(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, params ports.MergeParams) (domain.MergeRequest, error) {
1028 if _, err := m.writable(ctx, p, name, domain.ScopeMRMerge); err != nil {
1029 return domain.MergeRequest{}, err
1030 }
1031 mode := params.Mode
1032 if mode == "" {
1033 mode = domain.MergeModeFastForward
1034 }
1035 var v domain.Violations
1036 if _, err := domain.ParseMergeMode(mode.String()); err != nil {
1037 v.Add("mode", "is not a merge mode")
1038 }
1039 if params.HeadID.IsZero() {
1040 v.Add("head_id", "is required")
1041 }
1042 if err := v.Err(); err != nil {
1043 return domain.MergeRequest{}, err
1044 }
1045
1046 unlock, err := m.lockRepo(ctx, name)
1047 if err != nil {
1048 return domain.MergeRequest{}, err
1049 }
1050 var once sync.Once
1051 release := func() { once.Do(unlock) }
1052 defer release()
1053
1054 mr, err := m.store.Get(ctx, name, number)
1055 if err != nil {
1056 return domain.MergeRequest{}, err
1057 }
1058 // A merge moves the target and may delete the source: a token limited to
1059 // branch prefixes may do so only within them.
1060 if !p.AllowsRef("refs/heads/" + mr.Target) {
1061 return domain.MergeRequest{}, fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, mr.Target)
1062 }
1063 if params.DeleteSource && !p.AllowsRef("refs/heads/"+mr.Source) {
1064 return domain.MergeRequest{}, fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, mr.Source)
1065 }
1066 if !mr.IsOpen() {
1067 return domain.MergeRequest{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
1068 }
1069 if mr.IsDraft() {
1070 return domain.MergeRequest{}, fmt.Errorf("%w: !%d is a draft: mark it ready for review first", domain.ErrPreconditionFailed, number)
1071 }
1072 // The settings are read again under the lock: a change applies at once.
1073 repo, err := m.meta.Get(ctx, name)
1074 if err != nil {
1075 if errors.Is(err, domain.ErrNotFound) {
1076 return domain.MergeRequest{}, err // deleted meanwhile
1077 }
1078 return domain.MergeRequest{}, fmt.Errorf("get repository %q: %w", name, err)
1079 }
1080 settings := repo.EffectiveSettings()
1081 if !settings.Allows(mode) {
1082 return domain.MergeRequest{}, fmt.Errorf("%w: the %s merge mode is not allowed in %s", domain.ErrPreconditionFailed, mode, name)
1083 }
1084
1085 head, err := m.branchOrZero(ctx, name, mr.Source)
1086 switch {
1087 case err != nil:
1088 return domain.MergeRequest{}, err
1089 case head.IsZero():
1090 return domain.MergeRequest{}, fmt.Errorf("%w: the source branch %q no longer exists", domain.ErrPreconditionFailed, mr.Source)
1091 case head != params.HeadID:
1092 return domain.MergeRequest{}, fmt.Errorf("%w: branch moved, read again: %q is at %s, not %s", domain.ErrPreconditionFailed, mr.Source, head, params.HeadID)
1093 }
1094 tip, err := m.branchOrZero(ctx, name, mr.Target)
1095 switch {
1096 case err != nil:
1097 return domain.MergeRequest{}, err
1098 case tip.IsZero():
1099 return domain.MergeRequest{}, fmt.Errorf("%w: the target branch %q no longer exists", domain.ErrPreconditionFailed, mr.Target)
1100 case tip == head:
1101 return domain.MergeRequest{}, fmt.Errorf("%w: nothing to merge, %q already is at %s", domain.ErrPreconditionFailed, mr.Target, head)
1102 }
1103 // A head behind the target (or at it) has nothing to merge, in any mode.
1104 if behind, err := m.git.IsAncestor(ctx, name, head, tip); err != nil {
1105 return domain.MergeRequest{}, err
1106 } else if behind {
1107 return domain.MergeRequest{}, fmt.Errorf("%w: nothing to merge, %q already contains %s", domain.ErrPreconditionFailed, mr.Target, head)
1108 }
1109 tree, err := m.checkMode(ctx, mr, mode, tip, head)
1110 if err != nil {
1111 return domain.MergeRequest{}, err
1112 }
1113 if settings.FailingChecksBlock {
1114 if err := m.checksAllow(ctx, name, head); err != nil {
1115 return domain.MergeRequest{}, err
1116 }
1117 }
1118 if params.DeleteSource {
1119 if err := m.checkDeletable(ctx, name, mr); err != nil {
1120 return domain.MergeRequest{}, err
1121 }
1122 }
1123
1124 merged, err := m.moveTarget(ctx, mr, mode, tip, head, tree, p)
1125 if err != nil {
1126 return domain.MergeRequest{}, err
1127 }
1128 m.recordPush(ctx, name) // the target moved, whatever happens next
1129 mr, err = m.markMerged(ctx, name, number, mode, tip, head, merged)
1130 if err != nil {
1131 return domain.MergeRequest{}, err
1132 }
1133 log := zerowrap.FromCtx(ctx)
1134 log.Info().Str("repo", name.String()).Int64("number", number).Str("mode", mode.String()).
1135 Str("target", mr.Target).Str("commit", merged.String()).
1136 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request merged")
1137
1138 // The merge is done and recorded. The rest runs without the lock and after
1139 // the caller got its answer.
1140 release()
1141 bg := context.WithoutCancel(ctx)
1142 m.after.Add(1)
1143 go func() {
1144 defer m.after.Done()
1145 defer func() {
1146 if r := recover(); r != nil {
1147 log := zerowrap.FromCtx(bg)
1148 log.Error().Str("repo", name.String()).Int64("number", number).
1149 Str("panic", fmt.Sprint(r)).Bytes("stack", debug.Stack()).Msg("post-merge side effects panicked")
1150 }
1151 }()
1152 if m.afterGate != nil {
1153 m.afterGate()
1154 }
1155 m.afterMerge(bg, p, repo, mr, head, tip, merged, params.DeleteSource)
1156 }()
1157 return mr, nil
1158}
1159
1160// afterMerge is what follows a merge off the request path: errors are logged,
1161// never returned. mr is the merged request, tip the old target and merged the
1162// new one.
1163func (m *MergeRequests) afterMerge(ctx context.Context, p domain.Principal, repo domain.Repo, mr domain.MergeRequest, head, tip, merged domain.ObjectID, deleteSource bool) {
1164 name := repo.Name
1165 updates := []domain.RefUpdate{{Old: tip.String(), New: merged.String(), Ref: "refs/heads/" + mr.Target}}
1166 if deleteSource && m.deleteSource(ctx, p, name, mr, head) {
1167 updates = append(updates, domain.RefUpdate{Old: head.String(), New: domain.ZeroOID, Ref: "refs/heads/" + mr.Source})
1168 }
1169 at := mr.UpdatedAt
1170 recordActivity(ctx, m.meta, repo.Name, at)
1171 m.bus.Publish(ctx, domain.MergeRequestMerged{MergeRequest: mr, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
1172 m.bus.Publish(ctx, domain.PushEvent{
1173 Repo: name, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, Updates: updates,
1174 })
1175 if m.pulls != nil {
1176 m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: mr.Number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID, Merged: true}, p)
1177 }
1178 if m.ci != nil {
1179 // GitHub runs the "push" workflows of a merge, with the owner as actor;
1180 // the run is trusted with what the merging principal may read.
1181 m.ci.OnMerge(ctx, name, updates, p)
1182 }
1183 if m.mergeab != nil {
1184 m.mergeab.OnBranchesMoved(ctx, name, updates) // the other requests of the moved target
1185 }
1186}
1187
1188// deleteSource deletes the source branch of a merged request if it is still
1189// head and still unused, and reports whether it did. The eligibility is checked
1190// again and the branch deleted in one critical section of the merge lock: a
1191// request opened on the branch since the merge, or a push to it, keeps it. The
1192// record of who pushed the branch goes only after a deletion, and only if it
1193// still names head.
1194func (m *MergeRequests) deleteSource(ctx context.Context, p domain.Principal, name domain.RepoName, mr domain.MergeRequest, head domain.ObjectID) bool {
1195 log := zerowrap.FromCtx(ctx).With().Str("repo", name.String()).Int64("number", mr.Number).Str("branch", mr.Source).Logger()
1196 unlock, err := m.lockRepo(ctx, name)
1197 if err != nil {
1198 log.Warn().Err(err).Msg("source branch kept after the merge")
1199 return false
1200 }
1201 defer unlock()
1202 if err := m.checkDeletable(ctx, name, mr); err != nil {
1203 log.Warn().Err(err).Msg("source branch kept after the merge")
1204 return false
1205 }
1206 if err := m.git.DeleteBranch(ctx, name, mr.Source, head); err != nil {
1207 log.Warn().Err(err).Msg("source branch kept after the merge")
1208 return false
1209 }
1210 if m.tips != nil {
1211 if err := m.tips.DeleteRefTipIf(ctx, name, "refs/heads/"+mr.Source, head.String()); err != nil && !errors.Is(err, domain.ErrNotFound) {
1212 log.Warn().Err(err).Msg("the record of the source branch was kept")
1213 }
1214 }
1215 log.Info().Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("source branch deleted after the merge")
1216 return true
1217}
1218
1219// WhileOpen implements ports.PullGate: fn runs under the merge lock, so a
1220// close or a merge of the request cannot complete in the middle of it, and not
1221// at all when the request is no longer open.
1222func (m *MergeRequests) WhileOpen(ctx context.Context, name domain.RepoName, number int64, fn func() error) error {
1223 unlock, err := m.lockRepo(ctx, name)
1224 if err != nil {
1225 return err
1226 }
1227 defer unlock()
1228 mr, err := m.store.Get(ctx, name, number)
1229 if err != nil {
1230 return err
1231 }
1232 if !mr.IsOpen() {
1233 return fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
1234 }
1235 return fn()
1236}
1237
1238// checksAllow is the merge rule of FailingChecksBlock: every check of the
1239// head must have ended as success, skipped or neutral. A check still queued or
1240// running blocks too (the merge could land code that then fails), and a head
1241// without any check merges.
1242func (m *MergeRequests) checksAllow(ctx context.Context, name domain.RepoName, head domain.ObjectID) error {
1243 if m.checks == nil {
1244 return nil
1245 }
1246 checks, err := m.checks.CommitChecks(ctx, name, head.String())
1247 if errors.Is(err, domain.ErrResourceExhausted) {
1248 return fmt.Errorf("%w: too many checks to tell whether they passed: %v", domain.ErrPreconditionFailed, err)
1249 }
1250 if err != nil {
1251 return fmt.Errorf("read the checks of %s: %w", head, err)
1252 }
1253 var failing, pending []string
1254 for _, c := range checks {
1255 switch {
1256 case c.Status != domain.CICompleted:
1257 pending = append(pending, c.Name)
1258 case !c.Conclusion.PassesMerge():
1259 failing = append(failing, c.Name)
1260 }
1261 }
1262 switch {
1263 case len(failing) > 0:
1264 return fmt.Errorf("%w: failing checks block the merge: %s (turn off failing_checks_block to merge anyway)", domain.ErrPreconditionFailed, strings.Join(failing, ", "))
1265 case len(pending) > 0:
1266 return fmt.Errorf("%w: checks have not finished: %s", domain.ErrPreconditionFailed, strings.Join(pending, ", "))
1267 }
1268 return nil
1269}
1270
1271// checkDeletable refuses to delete the source branch of mr when it is the
1272// default branch or the source or target of another open request. It runs
1273// under the merge lock, which Create also holds while it checks its branches
1274// and stores the request.
1275func (m *MergeRequests) checkDeletable(ctx context.Context, name domain.RepoName, mr domain.MergeRequest) error {
1276 def, _, err := m.repos.DefaultBranch(ctx, name)
1277 if err != nil {
1278 return fmt.Errorf("default branch of %q: %w", name, err)
1279 }
1280 if mr.Source == def {
1281 return fmt.Errorf("%w: %q is the default branch of %s: it cannot be deleted", domain.ErrPreconditionFailed, mr.Source, name)
1282 }
1283 all, err := m.store.List(ctx, name)
1284 if err != nil {
1285 return fmt.Errorf("list merge requests of %q: %w", name, err)
1286 }
1287 for _, o := range all {
1288 if o.Number != mr.Number && o.IsOpen() && (o.Source == mr.Source || o.Target == mr.Source) {
1289 return fmt.Errorf("%w: !%d still uses %q: it cannot be deleted", domain.ErrPreconditionFailed, o.Number, mr.Source)
1290 }
1291 }
1292 return nil
1293}
1294
1295// recordPush does for a merge what Pushes.Record does for a push, as the merge
1296// moved a branch: it sets pushed_at and queues the maintenance of the
1297// repository. The target already moved, so a failure is logged, not returned,
1298// and the caller giving up does not stop it.
1299func (m *MergeRequests) recordPush(ctx context.Context, name domain.RepoName) {
1300 ctx = context.WithoutCancel(ctx)
1301 if _, err := touchPushedAt(ctx, m.meta, name, m.clock.Now().UTC()); err != nil {
1302 log := zerowrap.FromCtx(ctx)
1303 log.Error().Err(err).Str("repo", name.String()).Msg("record pushed_at after a merge")
1304 }
1305 m.maint.Schedule(ctx, name)
1306}
1307
1308// moveTarget moves the target branch from tip to the head (fast-forward) or to
1309// a new signed squash or merge commit with the given tree, and returns the new tip.
1310func (m *MergeRequests) moveTarget(ctx context.Context, mr domain.MergeRequest, mode domain.MergeMode, tip, head, tree domain.ObjectID, p domain.Principal) (domain.ObjectID, error) {
1311 newTip := head
1312 switch mode {
1313 case domain.MergeModeMerge:
1314 commit := domain.NewMergeCommit(mr, tree, tip, head, m.committer, m.clock.Now())
1315 armored, err := m.signer.Sign(commit.Payload())
1316 if err != nil {
1317 return domain.ObjectID{}, fmt.Errorf("sign the merge commit of !%d in %q: %w", mr.Number, mr.Repo, err)
1318 }
1319 if newTip, err = m.git.WriteCommit(ctx, mr.Repo, commit.Signed(armored)); err != nil {
1320 return domain.ObjectID{}, err
1321 }
1322 case domain.MergeModeSquash:
1323 commits, more, err := m.git.ListRange(ctx, mr.Repo, tip, head, domain.MaxSquashCommits)
1324 if err != nil {
1325 return domain.ObjectID{}, err
1326 }
1327 if more {
1328 return domain.ObjectID{}, fmt.Errorf("%w: the branch has more than %d commits to squash", domain.ErrPreconditionFailed, domain.MaxSquashCommits)
1329 }
1330 commit := domain.NewSquashCommit(mr, tree, tip, commits, m.committer, m.clock.Now())
1331 armored, err := m.signer.Sign(commit.Payload())
1332 if err != nil {
1333 return domain.ObjectID{}, fmt.Errorf("sign the squash commit of !%d in %q: %w", mr.Number, mr.Repo, err)
1334 }
1335 if newTip, err = m.git.WriteCommit(ctx, mr.Repo, commit.Signed(armored)); err != nil {
1336 return domain.ObjectID{}, err
1337 }
1338 }
1339 // The target is the merger's from the moment it moves: recorded before it
1340 // does, under the merge lock (domain.CIRefTip). The run shows the owner, as a
1341 // merge's push does on GitHub.
1342 if err := recordRefTips(ctx, m.tips, mr.Repo, []domain.RefUpdate{{Old: tip.String(), New: newTip.String(), Ref: "refs/heads/" + mr.Target}},
1343 domain.AuthorName(domain.Owner()), domain.PusherOf(p), domain.TrustedActor(p)); err != nil {
1344 return domain.ObjectID{}, err
1345 }
1346 err := m.git.UpdateBranch(ctx, mr.Repo, mr.Target, tip, newTip)
1347 if errors.Is(err, domain.ErrConflict) {
1348 return domain.ObjectID{}, fmt.Errorf("%w: the target branch %q moved during the merge, try again", domain.ErrPreconditionFailed, mr.Target)
1349 }
1350 if err != nil {
1351 return domain.ObjectID{}, err
1352 }
1353 return newTip, nil
1354}
1355
1356// markMerged records the merge. The target already moved, so a write that
1357// loses a race (a comment added meanwhile) is repeated on the fresh request.
1358func (m *MergeRequests) markMerged(ctx context.Context, name domain.RepoName, number int64, mode domain.MergeMode, base, head, merged domain.ObjectID) (domain.MergeRequest, error) {
1359 // The merge must be recorded even when the caller gives up.
1360 ctx = context.WithoutCancel(ctx)
1361 for range maxWriteRetries {
1362 mr, err := m.store.Get(ctx, name, number)
1363 if err != nil {
1364 return domain.MergeRequest{}, fmt.Errorf("record the merge of !%d in %q: %w", number, name, err)
1365 }
1366 mr.State, mr.HeadID, mr.BaseID, mr.MergedCommitID, mr.MergeMode = domain.MRStateMerged, head, base, merged, mode
1367 mr.UpdatedAt = m.clock.Now().UTC()
1368 err = m.store.Update(ctx, mr)
1369 if errors.Is(err, domain.ErrConflict) {
1370 continue
1371 }
1372 if err != nil {
1373 return domain.MergeRequest{}, fmt.Errorf("record the merge of !%d in %q: %w", number, name, err)
1374 }
1375 return mr, nil
1376 }
1377 log := zerowrap.FromCtx(ctx)
1378 log.Error().Str("repo", name.String()).Int64("number", number).Str("commit", merged.String()).
1379 Msg("the target branch moved but the merge request could not be marked merged")
1380 return domain.MergeRequest{}, fmt.Errorf("record the merge of !%d in %q: %w", number, name, domain.ErrConflict)
1381}
1382
1383// EditComment changes only the author's text, preserving the immutable verdict.
1384func (m *MergeRequests) EditComment(ctx context.Context, p domain.Principal, name domain.RepoName, number, commentID int64, text string) (domain.Comment, error) {
1385 return m.changeComment(ctx, p, name, number, commentID, text, false)
1386}
1387
1388// DeleteComment removes text but retains the anchor, ID and verdict. Repeated deletion is a no-op.
1389func (m *MergeRequests) DeleteComment(ctx context.Context, p domain.Principal, name domain.RepoName, number, commentID int64) (domain.Comment, error) {
1390 return m.changeComment(ctx, p, name, number, commentID, "", true)
1391}
1392
1393func (m *MergeRequests) changeComment(ctx context.Context, p domain.Principal, name domain.RepoName, number, commentID int64, text string, remove bool) (domain.Comment, error) {
1394 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
1395 if err != nil {
1396 return domain.Comment{}, err
1397 }
1398 for range maxWriteRetries {
1399 mr, err := m.store.Get(ctx, name, number)
1400 if err != nil {
1401 return domain.Comment{}, err
1402 }
1403 c, err := m.store.GetComment(ctx, name, number, commentID)
1404 if err != nil {
1405 return domain.Comment{}, err
1406 }
1407 if !mr.IsOpen() {
1408 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
1409 }
1410 if domain.AuthorName(p) != c.Author {
1411 return domain.Comment{}, fmt.Errorf("%w: only the author of a note can edit or delete it", domain.ErrForbidden)
1412 }
1413 if c.Deleted {
1414 if remove {
1415 return c, nil
1416 }
1417 return domain.Comment{}, fmt.Errorf("%w: the note was deleted", domain.ErrPreconditionFailed)
1418 }
1419 parsed, err := domain.ParsePlainText(text)
1420 var v domain.Violations
1421 if err != nil {
1422 v.AddErr("text", err)
1423 } else if !remove && parsed.IsBlank() && c.Verdict == domain.VerdictNone {
1424 v.Add("text", "is required")
1425 }
1426 if err := v.Err(); err != nil {
1427 return domain.Comment{}, err
1428 }
1429 if !remove && parsed == c.Text {
1430 return c, nil
1431 }
1432 c.Text, c.Deleted, c.EditedAt = parsed, remove, m.clock.Now().UTC()
1433 mr.UpdatedAt = c.EditedAt
1434 if err := m.store.SaveComment(ctx, mr, c); errors.Is(err, domain.ErrConflict) {
1435 continue
1436 } else if err != nil {
1437 return domain.Comment{}, fmt.Errorf("save note %d of !%d in %q: %w", commentID, number, name, err)
1438 }
1439 log := zerowrap.FromCtx(ctx)
1440 log.Info().Str("repo", name.String()).Int64("number", number).Int64("comment", c.ID).Bool("deleted", remove).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("comment changed")
1441 if remove {
1442 m.bus.Publish(ctx, domain.CommentDeleted{Repo: name, Number: number, Comment: c, At: c.EditedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
1443 } else {
1444 m.bus.Publish(ctx, domain.CommentEdited{Repo: name, Number: number, Comment: c, At: c.EditedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
1445 }
1446 return c, nil
1447 }
1448 return domain.Comment{}, fmt.Errorf("save note %d of !%d in %q: %w", commentID, number, name, domain.ErrConflict)
1449}
1450
1451// replyRoot returns the note a reply targets, or a violation when it is
1452// missing, itself a reply, or deleted.
1453func (m *MergeRequests) replyRoot(ctx context.Context, name domain.RepoName, number, id int64) (domain.Comment, error) {
1454 root, err := m.store.GetComment(ctx, name, number, id)
1455 var v domain.Violations
1456 switch {
1457 case errors.Is(err, domain.ErrNotFound):
1458 v.Add("reply_to", fmt.Sprintf("no note ^%d", id))
1459 case err != nil:
1460 return domain.Comment{}, err
1461 case root.IsReply() || root.Deleted:
1462 v.Add("reply_to", "must name a non-deleted root note")
1463 }
1464 if err := v.Err(); err != nil {
1465 return domain.Comment{}, err
1466 }
1467 return root, nil
1468}
1469
1470// Compare lists at most ports.MaxCompareCommits source commits absent from target. Unlike bounded
1471// history intersection, ancestry remains accurate even for deep histories.
1472func (m *MergeRequests) Compare(ctx context.Context, p domain.Principal, repo domain.RepoName, target, source domain.ObjectID) (ports.CompareResult, error) {
1473 if _, err := m.readable(ctx, p, repo); err != nil {
1474 return ports.CompareResult{}, err
1475 }
1476 if target == source {
1477 return ports.CompareResult{Exact: true}, nil
1478 }
1479 ancestor, err := m.git.IsAncestor(ctx, repo, target, source)
1480 if err != nil {
1481 return ports.CompareResult{}, err
1482 }
1483 commits, more, err := m.git.ListRange(ctx, repo, target, source, ports.MaxCompareCommits)
1484 if err != nil {
1485 return ports.CompareResult{}, err
1486 }
1487 verifyCommits(ctx, m.verifier, repo, commits)
1488 return ports.CompareResult{Ahead: commits, Behind: !ancestor, Exact: !more}, nil
1489}
1490
1491// PushSummary detects rewrites by ancestry, not log ordering, and counts at
1492// most ports.MaxPushSummaryCommits new commits. More distinguishes a longer range from a rewrite.
1493func (m *MergeRequests) PushSummary(ctx context.Context, p domain.Principal, repo domain.RepoName, oldHead, newTip domain.ObjectID) (ports.PushSummary, error) {
1494 if _, err := m.readable(ctx, p, repo); err != nil {
1495 return ports.PushSummary{}, err
1496 }
1497 if oldHead == newTip {
1498 return ports.PushSummary{}, nil
1499 }
1500 ancestor, err := m.git.IsAncestor(ctx, repo, oldHead, newTip)
1501 if err != nil {
1502 return ports.PushSummary{}, err
1503 }
1504 if !ancestor {
1505 return ports.PushSummary{Rewritten: true}, nil
1506 }
1507 commits, more, err := m.git.ListRange(ctx, repo, oldHead, newTip, ports.MaxPushSummaryCommits)
1508 if err != nil {
1509 return ports.PushSummary{}, err
1510 }
1511 return ports.PushSummary{Count: len(commits), More: more}, nil
1512}