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