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