1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "hash/fnv"
10 "net/url"
11 "slices"
12 "strconv"
13 "strings"
14 "sync"
15 "unicode/utf8"
16
17 kv "bnema.dev/ermite-kv"
18 "github.com/bnema/zerowrap"
19 "google.golang.org/protobuf/proto"
20 "google.golang.org/protobuf/types/known/timestamppb"
21
22 storagepb "bnema.dev/ermite/gen/ermite/storage"
23 "bnema.dev/ermite/internal/domain"
24 "bnema.dev/ermite/internal/ports"
25)
26
27// Key layout of the merge requests. Repository names hold no '/', so a prefix
28// ending in "/" never matches another repository; numbers and comment ids are
29// zero-padded so that keys sort numerically.
30const (
31 mrPrefix = "mr/" // mr/<repo>/<number:010d>
32 commentPrefix = "mrc/" // mrc/<repo>/<number:010d>/<id:010d>
33 readPrefix = "mrread/" // mrread/<repo>/<number:010d>
34 seqPrefix = "mrseq/" // mrseq/<repo>
35 mergePrefix = "mrmerge/" // mrmerge/<repo>/<number:010d>
36 // The indexes of the open merge requests by source branch and by target
37 // branch, which a push looks up (the values are not read). They are written in
38 // the same batch as the merge request, so they are never out of step with it,
39 // and built once, together, for the data that predates them (BuildOpenIndex,
40 // mrOpenIndexMarker).
41 mrOpenPrefix = "mropen/" // mropen/<repo>/<escaped source>/<number:010d>
42 mrTargetPrefix = "mrtarget/" // mrtarget/<repo>/<escaped target>/<number:010d>
43 mrOpenIndexMarker = "meta/mropen" // present once the indexes cover every merge request
44 numberFormat = "%010d"
45 createTries = 10
46)
47
48// MergeRequests is the ermite-kv backed ports.MergeRequestStore. It shares the
49// database of the Store it comes from.
50type MergeRequests struct {
51 db *kv.DB
52 locks *numberLocks
53}
54
55var _ ports.MergeRequestStore = (*MergeRequests)(nil)
56
57// MergeRequests returns the merge request store over the same database. It is
58// valid until the Store is closed.
59func (s *Store) MergeRequests() *MergeRequests { return &MergeRequests{db: s.db, locks: s.locks} }
60
61func mrRepoPrefix(repo domain.RepoName) string { return mrPrefix + repo.String() + "/" }
62func mrKey(repo domain.RepoName, n int64) []byte {
63 return []byte(mrRepoPrefix(repo) + fmt.Sprintf(numberFormat, n))
64}
65func commentsPrefix(repo domain.RepoName, n int64) string {
66 return commentPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, n) + "/"
67}
68func commentKey(repo domain.RepoName, n, id int64) []byte {
69 return []byte(commentsPrefix(repo, n) + fmt.Sprintf(numberFormat, id))
70}
71func readKey(repo domain.RepoName, n int64) []byte {
72 return []byte(readPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, n))
73}
74func mergeKey(repo domain.RepoName, n int64) []byte {
75 return []byte(mergePrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, n))
76}
77func seqKey(repo domain.RepoName) []byte { return []byte(seqPrefix + repo.String()) }
78
79// mrOpenRepoPrefix is where the open merge requests of a repository are indexed
80// by source branch.
81func mrOpenRepoPrefix(repo domain.RepoName) string { return mrOpenPrefix + repo.String() + "/" }
82
83// indexBranchPrefix is where the open merge requests of one branch are indexed
84// under prefix (mrOpenPrefix or mrTargetPrefix): the branch is escaped, so it
85// holds no '/' and the prefix matches that branch only.
86func indexBranchPrefix(prefix string, repo domain.RepoName, branch string) string {
87 return prefix + repo.String() + "/" + url.PathEscape(branch) + "/"
88}
89
90func indexKey(prefix string, repo domain.RepoName, branch string, n int64) []byte {
91 return []byte(indexBranchPrefix(prefix, repo, branch) + fmt.Sprintf(numberFormat, n))
92}
93
94// indexOpen adds to b the writes that keep the indexes of the open merge
95// requests (by source and by target) in step with a write of mr over old (nil
96// when old is not known, as for a new request or a comment: the keys of mr are
97// then put or deleted, whatever they were). Deleting a key that is not there is
98// not an error.
99//
100//hexcheck:ignore suspicious-business-logic-in-adapter maintains source and target indexes to match the stored merge request open state and branch names
101func indexOpen(b *kv.Batch, old *domain.MergeRequest, mr domain.MergeRequest) {
102 if old != nil && old.IsOpen() {
103 if !mr.IsOpen() || old.Source != mr.Source {
104 b.Delete(indexKey(mrOpenPrefix, old.Repo, old.Source, old.Number))
105 }
106 if !mr.IsOpen() || old.Target != mr.Target {
107 b.Delete(indexKey(mrTargetPrefix, old.Repo, old.Target, old.Number))
108 }
109 }
110 switch {
111 case mr.IsOpen():
112 b.Put(indexKey(mrOpenPrefix, mr.Repo, mr.Source, mr.Number), []byte{1})
113 b.Put(indexKey(mrTargetPrefix, mr.Repo, mr.Target, mr.Number), []byte{1})
114 case old == nil:
115 b.Delete(indexKey(mrOpenPrefix, mr.Repo, mr.Source, mr.Number))
116 b.Delete(indexKey(mrTargetPrefix, mr.Repo, mr.Target, mr.Number))
117 }
118}
119
120// BuildOpenIndex builds the indexes of the open merge requests (by source and
121// by target) when the data does not have them yet (it predates them): one pass
122// over the merge requests, in batches, then the marker. It runs when the store is opened and is idempotent: a pass
123// cut short by a crash is simply done again, writing the same keys.
124//
125//hexcheck:ignore suspicious-business-logic-in-adapter one-time storage backfill of an index: a storage concern, no policy
126func (s *Store) BuildOpenIndex(ctx context.Context) error {
127 if _, ok := s.db.Get(ctx, []byte(mrOpenIndexMarker)); ok {
128 return nil
129 }
130 var b kv.Batch
131 n := 0
132 flush := func() error {
133 if b.Len() == 0 {
134 return nil
135 }
136 if _, err := s.db.Apply(ctx, &b); err != nil {
137 return fmt.Errorf("build the index of open merge requests: %w", err)
138 }
139 b.Reset()
140 return nil
141 }
142 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(mrPrefix)) {
143 repoName, _, ok := strings.Cut(strings.TrimPrefix(string(key), mrPrefix), "/")
144 if !ok {
145 continue
146 }
147 repo, err := domain.ParseRepoName(repoName)
148 if err != nil {
149 continue
150 }
151 mr, err := decodeMergeRequest(repo, item)
152 if err != nil {
153 return fmt.Errorf("build the index of open merge requests: decode %s: %w", key, err)
154 }
155 if !mr.IsOpen() {
156 continue
157 }
158 b.Put(indexKey(mrOpenPrefix, mr.Repo, mr.Source, mr.Number), []byte{1})
159 b.Put(indexKey(mrTargetPrefix, mr.Repo, mr.Target, mr.Number), []byte{1})
160 if n++; n%(deleteChunk/2) == 0 {
161 if err := flush(); err != nil {
162 return err
163 }
164 }
165 }
166 if err := flush(); err != nil {
167 return err
168 }
169 b.Put([]byte(mrOpenIndexMarker), []byte{1})
170 if _, err := s.db.Apply(ctx, &b); err != nil {
171 return fmt.Errorf("build the index of open merge requests: %w", err)
172 }
173 log := zerowrap.FromCtx(ctx)
174 log.Info().Int("open", n).Msg("index of open merge requests built")
175 return nil
176}
177
178// mergeRequestKeys lists every key of the snapshot that belongs to the
179// repository's merge requests.
180func mergeRequestKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
181 var keys [][]byte
182 for _, p := range []string{
183 mrRepoPrefix(repo), commentPrefix + repo.String() + "/", readPrefix + repo.String() + "/",
184 mrOpenRepoPrefix(repo), mrTargetPrefix + repo.String() + "/", mrRefPrefix + repo.String() + "/",
185 mergePrefix + repo.String() + "/",
186 } {
187 for k := range snap.Prefix(ctx, []byte(p)) {
188 keys = append(keys, slices.Clone(k))
189 }
190 }
191 if _, ok := snap.Get(ctx, seqKey(repo)); ok {
192 keys = append(keys, seqKey(repo))
193 }
194 return keys
195}
196
197// guardRepo is guardRepoKey on the database of the merge requests; see there.
198func (m *MergeRequests) guardRepo(ctx context.Context, b *kv.Batch, repo domain.RepoName) error {
199 return guardRepoKey(ctx, m.db, b, repo)
200}
201
202// Create takes the next number of the repository from mrseq/<repo> and stores
203// the merge request under it, in one batch that requires the sequence to be
204// unchanged since it was read: two concurrent creations never share a number.
205// The batch also requires the repository to be unchanged since it was read and
206// rewrites it (see guardRepo), so that a repository deleted meanwhile is not
207// left with a stray merge request (a push recording pushed_at makes it retry). A missing repository gives
208// domain.ErrNotFound.
209func (m *MergeRequests) Create(ctx context.Context, mr domain.MergeRequest) (domain.MergeRequest, error) {
210 err := createNumbered(ctx, m.db, m.locks, mr.Repo, func(b *kv.Batch, n int64) error {
211 mr.Number = n
212 val, err := encodeMergeRequest(mr)
213 if err != nil {
214 return err
215 }
216 b.RequireAbsent(mrKey(mr.Repo, n))
217 b.Put(mrKey(mr.Repo, n), val)
218 indexOpen(b, nil, mr)
219 return nil
220 })
221 if err != nil {
222 return domain.MergeRequest{}, fmt.Errorf("create merge request of %q: %w", mr.Repo, err)
223 }
224 return m.Get(ctx, mr.Repo, mr.Number)
225}
226
227// createNumbered takes the next number of the repository from mrseq/<repo>,
228// shared by the merge requests and the issues, and lets add put the record
229// under it into a batch that requires the sequence to be unchanged since it was
230// read (absent for the first number) and the repository to be unchanged (see
231// guardRepoKey): two concurrent creations, of either kind, never share a
232// number. It retries when the batch conflicts; a missing repository gives
233// domain.ErrNotFound.
234//
235// Creations in one repository are serialized in the process (numberLocks), so
236// a burst of merge requests and issues never exhausts the retries; the version
237// checks still guard against any other writer.
238func createNumbered(ctx context.Context, db *kv.DB, locks *numberLocks, repo domain.RepoName, add func(b *kv.Batch, number int64) error) error {
239 unlock := locks.lock(repo)
240 defer unlock()
241 for range createTries {
242 repoItem, repoOK := db.Get(ctx, repoKey(repo))
243 if !repoOK {
244 return domain.ErrNotFound
245 }
246 var last int64
247 item, ok := db.Get(ctx, seqKey(repo))
248 if ok {
249 var seq storagepb.Sequence
250 if err := proto.Unmarshal(item.Value, &seq); err != nil {
251 return fmt.Errorf("decode sequence of %q: %w", repo, err)
252 }
253 last = seq.GetLast()
254 }
255 number := last + 1
256 var b kv.Batch
257 b.RequireVersion(repoKey(repo), repoItem.Version)
258 b.Put(repoKey(repo), repoItem.Value) // see guardRepoKey
259 if ok {
260 b.RequireVersion(seqKey(repo), item.Version)
261 } else {
262 b.RequireAbsent(seqKey(repo))
263 }
264 b.Put(seqKey(repo), mustMarshal(&storagepb.Sequence{Last: number}))
265 if err := add(&b, number); err != nil {
266 return err
267 }
268 _, err := db.Apply(ctx, &b)
269 if errors.Is(err, kv.ErrConflict) {
270 continue
271 }
272 return err
273 }
274 return domain.ErrConflict
275}
276
277// numberLockStripes is the size of the table of numbering locks.
278const numberLockStripes = 64
279
280// numberLocks is a fixed table of mutexes, one stripe per hash of the
281// repository name, for createNumbered: the memory is bounded whatever the
282// number of repositories, and two repositories sharing a stripe only wait for
283// each other. One table is shared by every store of a Store, so the merge
284// requests and the issues of a repository are numbered one at a time.
285type numberLocks [numberLockStripes]sync.Mutex
286
287// lock locks the numbering of repo and returns the unlock function.
288func (l *numberLocks) lock(repo domain.RepoName) func() {
289 h := fnv.New32a()
290 _, _ = h.Write([]byte(repo.String()))
291 mu := &l[h.Sum32()%numberLockStripes]
292 mu.Lock()
293 return mu.Unlock
294}
295
296// Get reads mr/<repo>/<number>.
297func (m *MergeRequests) Get(ctx context.Context, repo domain.RepoName, number int64) (domain.MergeRequest, error) {
298 if number < 1 {
299 return domain.MergeRequest{}, domain.ErrNotFound
300 }
301 item, ok := m.db.Get(ctx, mrKey(repo, number))
302 if !ok {
303 return domain.MergeRequest{}, domain.ErrNotFound
304 }
305 return decodeMergeRequest(repo, item)
306}
307
308// List returns the merge requests of the repository, highest number first.
309func (m *MergeRequests) List(ctx context.Context, repo domain.RepoName) ([]domain.MergeRequest, error) {
310 var out []domain.MergeRequest
311 for key, item := range m.db.Snapshot(ctx).Prefix(ctx, []byte(mrRepoPrefix(repo))) {
312 mr, err := decodeMergeRequest(repo, item)
313 if err != nil {
314 return nil, fmt.Errorf("decode %s: %w", key, err)
315 }
316 out = append(out, mr)
317 }
318 slices.Reverse(out)
319 return out, nil
320}
321
322// scanIndex reads the numbers under the prefixes of an index, in key order. It
323// reads keys only: no merge request is decoded. With a limit above 0 it stops as
324// soon as it has seen limit+1 keys and returns the first limit with truncated
325// set, so the work does not grow with the number of open requests on a branch.
326// The context is checked before every prefix and every key (the scan of ermite-kv
327// itself ignores it): on cancellation it stops and returns what it has with the
328// error of the context.
329func (m *MergeRequests) scanIndex(ctx context.Context, prefixes []string, limit int) (numbers []int64, truncated bool, err error) {
330 snap := m.db.Snapshot(ctx)
331 for _, prefix := range prefixes {
332 if err := ctx.Err(); err != nil {
333 return numbers, true, err
334 }
335 for key := range snap.Prefix(ctx, []byte(prefix)) {
336 if err := ctx.Err(); err != nil {
337 return numbers, true, err
338 }
339 if limit > 0 && len(numbers) == limit {
340 return numbers, true, nil // the limit+1th key: more exist
341 }
342 n, perr := strconv.ParseInt(string(key[strings.LastIndexByte(string(key), '/')+1:]), 10, 64)
343 if perr != nil {
344 return numbers, false, fmt.Errorf("index key %q: %w", key, perr)
345 }
346 numbers = append(numbers, n)
347 }
348 }
349 return numbers, false, nil
350}
351
352func (m *MergeRequests) byBranch(ctx context.Context, index string, repo domain.RepoName, branches []string, limit int) ([]int64, bool, error) {
353 prefixes := make([]string, 0, len(branches))
354 for _, b := range branches {
355 prefixes = append(prefixes, indexBranchPrefix(index, repo, b))
356 }
357 return m.scanIndex(ctx, prefixes, limit)
358}
359
360// OpenBySource returns the numbers of the open merge requests of the repository
361// whose source branch is one of branches, from the index: a scan of the keys of
362// those branches only, whatever the number of merge requests the repository
363// ever had, stopping after limit of them (truncated).
364func (m *MergeRequests) OpenBySource(ctx context.Context, repo domain.RepoName, branches []string, limit int) ([]int64, bool, error) {
365 return m.byBranch(ctx, mrOpenPrefix, repo, branches, limit)
366}
367
368// OpenByTarget is OpenBySource for the target branch.
369func (m *MergeRequests) OpenByTarget(ctx context.Context, repo domain.RepoName, branches []string, limit int) ([]int64, bool, error) {
370 return m.byBranch(ctx, mrTargetPrefix, repo, branches, limit)
371}
372
373// ListOpen returns the open merge requests of the repository, highest number
374// first, through the index: only the open ones are read.
375//
376//hexcheck:ignore suspicious-business-logic-in-adapter reads the open merge requests through their index: a storage concern, no policy
377func (m *MergeRequests) ListOpen(ctx context.Context, repo domain.RepoName) ([]domain.MergeRequest, error) {
378 numbers, _, err := m.scanIndex(ctx, []string{mrOpenRepoPrefix(repo)}, 0)
379 if err != nil {
380 return nil, err
381 }
382 slices.Sort(numbers)
383 var out []domain.MergeRequest
384 for _, n := range slices.Backward(numbers) {
385 mr, err := m.Get(ctx, repo, n)
386 if errors.Is(err, domain.ErrNotFound) {
387 continue // closed or deleted since the snapshot of the index
388 }
389 if err != nil {
390 return nil, err
391 }
392 if mr.IsOpen() {
393 out = append(out, mr)
394 }
395 }
396 return out, nil
397}
398
399// ListOpenAll returns the open merge requests of every repository, repository
400// by repository and highest number first, through the mropen/ index: the
401// requests it names are read, no other.
402//
403//hexcheck:ignore suspicious-business-logic-in-adapter reads the open merge requests of all repositories through their index: a storage concern, no policy
404func (m *MergeRequests) ListOpenAll(ctx context.Context) ([]domain.MergeRequest, error) {
405 type ref struct {
406 repo domain.RepoName
407 n int64
408 }
409 var refs []ref
410 for key := range m.db.Snapshot(ctx).Prefix(ctx, []byte(mrOpenPrefix)) {
411 if err := ctx.Err(); err != nil {
412 return nil, err
413 }
414 rest := strings.TrimPrefix(string(key), mrOpenPrefix)
415 repoName, _, ok := strings.Cut(rest, "/")
416 if !ok {
417 return nil, fmt.Errorf("index key %q", key)
418 }
419 repo, err := domain.ParseRepoName(repoName)
420 if err != nil {
421 return nil, fmt.Errorf("index key %q: %w", key, err)
422 }
423 n, err := strconv.ParseInt(rest[strings.LastIndexByte(rest, '/')+1:], 10, 64)
424 if err != nil {
425 return nil, fmt.Errorf("index key %q: %w", key, err)
426 }
427 refs = append(refs, ref{repo, n})
428 }
429 var out []domain.MergeRequest
430 for _, r := range refs {
431 mr, err := m.Get(ctx, r.repo, r.n)
432 if errors.Is(err, domain.ErrNotFound) {
433 continue // closed or deleted since the snapshot of the index
434 }
435 if err != nil {
436 return nil, err
437 }
438 if mr.IsOpen() {
439 out = append(out, mr)
440 }
441 }
442 return out, nil
443}
444
445// CountOpen counts the open merge requests of the repository from the keys of
446// the index: no request is decoded.
447func (m *MergeRequests) CountOpen(ctx context.Context, repo domain.RepoName) (int, error) {
448 numbers, _, err := m.scanIndex(ctx, []string{mrOpenRepoPrefix(repo)}, 0)
449 return len(numbers), err
450}
451
452// Update replaces the merge request with a version precondition.
453//
454//hexcheck:ignore suspicious-business-logic-in-adapter replaces a versioned record and maintains open-request indexes and the derived mergeability record
455func (m *MergeRequests) Update(ctx context.Context, mr domain.MergeRequest) error {
456 val, err := encodeMergeRequest(mr)
457 if err != nil {
458 return err
459 }
460 key := mrKey(mr.Repo, mr.Number)
461 cur, ok := m.db.Get(ctx, key)
462 if !ok {
463 return domain.ErrNotFound
464 }
465 if mr.Version == 0 {
466 return fmt.Errorf("update merge request %d of %q: read it first: %w", mr.Number, mr.Repo, domain.ErrConflict)
467 }
468 var b kv.Batch
469 b.RequireVersion(key, mr.Version)
470 b.Put(key, val)
471 // The stored request is the one mr was read at (the version is required
472 // above): its index key moves with the write.
473 if old, err := decodeMergeRequest(mr.Repo, cur); err == nil {
474 indexOpen(&b, &old, mr)
475 } else {
476 indexOpen(&b, nil, mr)
477 }
478 if !mr.IsOpen() {
479 b.Delete(mergeKey(mr.Repo, mr.Number))
480 }
481 if _, err := m.db.Apply(ctx, &b); err != nil {
482 if errors.Is(err, kv.ErrConflict) {
483 return domain.ErrConflict
484 }
485 return fmt.Errorf("update merge request %d of %q: %w", mr.Number, mr.Repo, err)
486 }
487 return nil
488}
489
490// ListComments returns the comments of the merge request, oldest first.
491func (m *MergeRequests) ListComments(ctx context.Context, repo domain.RepoName, number int64) ([]domain.Comment, error) {
492 var out []domain.Comment
493 for key, item := range m.db.Snapshot(ctx).Prefix(ctx, []byte(commentsPrefix(repo, number))) {
494 c, err := decodeComment(item)
495 if err != nil {
496 return nil, fmt.Errorf("decode %s: %w", key, err)
497 }
498 out = append(out, c)
499 }
500 return out, nil
501}
502
503// GetComment reads one comment.
504func (m *MergeRequests) GetComment(ctx context.Context, repo domain.RepoName, number, id int64) (domain.Comment, error) {
505 if number < 1 || id < 1 {
506 return domain.Comment{}, domain.ErrNotFound
507 }
508 item, ok := m.db.Get(ctx, commentKey(repo, number, id))
509 if !ok {
510 return domain.Comment{}, domain.ErrNotFound
511 }
512 return decodeComment(item)
513}
514
515// SaveComment writes the comment and the merge request in one batch, each with
516// its version precondition (absent for a new comment), and requires the
517// repository to exist (see guardRepo): domain.ErrNotFound otherwise.
518func (m *MergeRequests) SaveComment(ctx context.Context, mr domain.MergeRequest, c domain.Comment) error {
519 mrVal, err := encodeMergeRequest(mr)
520 if err != nil {
521 return err
522 }
523 cVal, err := encodeComment(c)
524 if err != nil {
525 return err
526 }
527 if mr.Version == 0 {
528 return fmt.Errorf("save comment on %d of %q: read the merge request first: %w", mr.Number, mr.Repo, domain.ErrConflict)
529 }
530 var b kv.Batch
531 if err := m.guardRepo(ctx, &b, mr.Repo); err != nil {
532 return err
533 }
534 b.RequireVersion(mrKey(mr.Repo, mr.Number), mr.Version)
535 b.Put(mrKey(mr.Repo, mr.Number), mrVal)
536 indexOpen(&b, nil, mr)
537 if c.Version == 0 {
538 b.RequireAbsent(commentKey(mr.Repo, mr.Number, c.ID))
539 } else {
540 b.RequireVersion(commentKey(mr.Repo, mr.Number, c.ID), c.Version)
541 }
542 b.Put(commentKey(mr.Repo, mr.Number, c.ID), cVal)
543 if _, err := m.db.Apply(ctx, &b); err != nil {
544 if errors.Is(err, kv.ErrConflict) {
545 return domain.ErrConflict
546 }
547 return fmt.Errorf("save comment on %d of %q: %w", mr.Number, mr.Repo, err)
548 }
549 return nil
550}
551
552// GetRead reads mrread/<repo>/<number>: the id of a commit, or nothing.
553func (m *MergeRequests) GetRead(ctx context.Context, repo domain.RepoName, number int64) (domain.ObjectID, error) {
554 item, ok := m.db.Get(ctx, readKey(repo, number))
555 if !ok {
556 return domain.ObjectID{}, nil
557 }
558 id, err := domain.ParseObjectID(string(item.Value))
559 if err != nil {
560 return domain.ObjectID{}, fmt.Errorf("decode read position of %d in %q: %w", number, repo, err)
561 }
562 return id, nil
563}
564
565// SetRead writes mrread/<repo>/<number>. The repository must exist (see
566// guardRepo): domain.ErrNotFound otherwise.
567func (m *MergeRequests) SetRead(ctx context.Context, repo domain.RepoName, number int64, head domain.ObjectID) error {
568 if head.IsZero() {
569 return fmt.Errorf("%w: empty read position", domain.ErrInvalidObjectID)
570 }
571 for range createTries {
572 var b kv.Batch
573 if err := m.guardRepo(ctx, &b, repo); err != nil {
574 return err
575 }
576 b.Put(readKey(repo, number), []byte(head.String()))
577 _, err := m.db.Apply(ctx, &b)
578 if errors.Is(err, kv.ErrConflict) {
579 continue
580 }
581 if err != nil {
582 return fmt.Errorf("record read position of %d in %q: %w", number, repo, err)
583 }
584 return nil
585 }
586 return fmt.Errorf("record read position of %d in %q: %w", number, repo, domain.ErrConflict)
587}
588
589// GetMergeability returns the stored mergeability, or domain.ErrNotFound.
590func (m *MergeRequests) GetMergeability(ctx context.Context, repo domain.RepoName, number int64) (domain.Mergeability, error) {
591 item, ok := m.db.Get(ctx, mergeKey(repo, number))
592 if !ok {
593 return domain.Mergeability{}, domain.ErrNotFound
594 }
595 return decodeMergeability(repo, number, item)
596}
597
598// PutMergeability stores the mergeability in a batch that requires the merge
599// request to be unchanged since it was read here and open, and the record to
600// be absent (Version 0) or unchanged.
601//
602//hexcheck:ignore suspicious-business-logic-in-adapter version preconditions of a derived record: a storage concern, no policy
603func (m *MergeRequests) PutMergeability(ctx context.Context, mg domain.Mergeability) error {
604 val, err := encodeMergeability(mg)
605 if err != nil {
606 return err
607 }
608 for range createTries {
609 mrItem, ok := m.db.Get(ctx, mrKey(mg.Repo, mg.Number))
610 if !ok {
611 return domain.ErrNotFound
612 }
613 mr, err := decodeMergeRequest(mg.Repo, mrItem)
614 if err != nil {
615 return err
616 }
617 if !mr.IsOpen() {
618 return domain.ErrNotFound
619 }
620 key := mergeKey(mg.Repo, mg.Number)
621 var b kv.Batch
622 b.RequireVersion(mrKey(mg.Repo, mg.Number), mrItem.Version)
623 if mg.Version == 0 {
624 b.RequireAbsent(key)
625 } else {
626 b.RequireVersion(key, mg.Version)
627 }
628 b.Put(key, val)
629 _, err = m.db.Apply(ctx, &b)
630 if errors.Is(err, kv.ErrConflict) {
631 // Either the record changed (the caller's conflict) or the merge
632 // request did (retry against it).
633 if cur, ok := m.db.Get(ctx, mrKey(mg.Repo, mg.Number)); ok && cur.Version != mrItem.Version {
634 continue
635 }
636 return domain.ErrConflict
637 }
638 if err != nil {
639 return fmt.Errorf("store mergeability of %d in %q: %w", mg.Number, mg.Repo, err)
640 }
641 return nil
642 }
643 return fmt.Errorf("store mergeability of %d in %q: %w", mg.Number, mg.Repo, domain.ErrConflict)
644}
645
646// DeleteMergeability removes the stored mergeability; absent is fine.
647func (m *MergeRequests) DeleteMergeability(ctx context.Context, repo domain.RepoName, number int64) error {
648 var b kv.Batch
649 b.Delete(mergeKey(repo, number))
650 if _, err := m.db.Apply(ctx, &b); err != nil {
651 return fmt.Errorf("delete mergeability of %d in %q: %w", number, repo, err)
652 }
653 return nil
654}
655
656func encodeMergeability(m domain.Mergeability) ([]byte, error) {
657 var state storagepb.MergeabilityState
658 switch m.State {
659 case domain.MergeabilityUpToDate:
660 state = storagepb.MergeabilityState_MERGEABILITY_STATE_UP_TO_DATE
661 case domain.MergeabilityBehind:
662 state = storagepb.MergeabilityState_MERGEABILITY_STATE_BEHIND
663 case domain.MergeabilityConflict:
664 state = storagepb.MergeabilityState_MERGEABILITY_STATE_CONFLICT
665 default:
666 return nil, fmt.Errorf("%w: mergeability state %q is not stored", domain.ErrInvalid, m.State)
667 }
668 return proto.Marshal(&storagepb.Mergeability{
669 State: state,
670 TargetId: m.TargetID.String(),
671 HeadId: m.HeadID.String(),
672 Behind: int32(m.Behind), //nolint:gosec // capped at MaxBehindCount
673 BehindCapped: m.BehindCapped,
674 ConflictPaths: m.ConflictPaths,
675 ConflictPathsTruncated: m.ConflictPathsTruncated,
676 CheckedAt: timestamppb.New(m.CheckedAt),
677 })
678}
679
680func decodeMergeability(repo domain.RepoName, number int64, item kv.Item) (domain.Mergeability, error) {
681 var pb storagepb.Mergeability
682 if err := proto.Unmarshal(item.Value, &pb); err != nil {
683 return domain.Mergeability{}, fmt.Errorf("decode mergeability of %d in %q: %w", number, repo, err)
684 }
685 state, ok := mergeabilityStates[pb.GetState()]
686 if !ok {
687 return domain.Mergeability{}, fmt.Errorf("decode mergeability of %d in %q: unknown state %d: %w", number, repo, pb.GetState(), domain.ErrNotFound)
688 }
689 target, err := domain.ParseObjectID(pb.GetTargetId())
690 if err != nil {
691 return domain.Mergeability{}, fmt.Errorf("decode mergeability of %d in %q: %w", number, repo, err)
692 }
693 head, err := domain.ParseObjectID(pb.GetHeadId())
694 if err != nil {
695 return domain.Mergeability{}, fmt.Errorf("decode mergeability of %d in %q: %w", number, repo, err)
696 }
697 return domain.Mergeability{
698 Repo: repo, Number: number, State: state,
699 TargetID: target, HeadID: head,
700 Behind: int(pb.GetBehind()), BehindCapped: pb.GetBehindCapped(),
701 ConflictPaths: pb.GetConflictPaths(),
702 ConflictPathsTruncated: pb.GetConflictPathsTruncated(),
703 CheckedAt: pb.GetCheckedAt().AsTime(),
704 Version: item.Version,
705 }, nil
706}
707
708// mergeabilityStates maps the stored states to the domain ones; CHECKING is
709// never stored.
710var mergeabilityStates = map[storagepb.MergeabilityState]domain.MergeabilityState{
711 storagepb.MergeabilityState_MERGEABILITY_STATE_UP_TO_DATE: domain.MergeabilityUpToDate,
712 storagepb.MergeabilityState_MERGEABILITY_STATE_BEHIND: domain.MergeabilityBehind,
713 storagepb.MergeabilityState_MERGEABILITY_STATE_CONFLICT: domain.MergeabilityConflict,
714}
715
716func encodeMergeRequest(mr domain.MergeRequest) ([]byte, error) {
717 if mr.Repo.IsZero() {
718 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
719 }
720 if !utf8.ValidString(mr.Author) {
721 return nil, errors.New("merge request author is not valid UTF-8")
722 }
723 state, err := encodeState(mr.State)
724 if err != nil {
725 return nil, err
726 }
727 mode, err := encodeMode(mr.MergeMode)
728 if err != nil {
729 return nil, err
730 }
731 var autoMerge *storagepb.AutoMerge
732 if am := mr.AutoMerge; am != nil {
733 amMode, err := encodeMode(am.Mode)
734 if err != nil || amMode == storagepb.MergeMode_MERGE_MODE_UNSPECIFIED {
735 return nil, fmt.Errorf("%w: merge mode %q in the auto-merge", domain.ErrInvalid, am.Mode)
736 }
737 autoMerge = &storagepb.AutoMerge{
738 Mode: amMode, HeadId: am.HeadID.String(),
739 EnablerKind: string(am.EnabledBy.Kind), EnablerTokenId: am.EnabledBy.TokenID,
740 EnablerLabel: am.EnablerLabel, EnabledAt: encodeTime(am.EnabledAt),
741 }
742 }
743 return mustMarshal(&storagepb.MergeRequest{
744 Number: mr.Number,
745 Title: mr.Title.String(),
746 Description: mr.Description.String(),
747 SourceBranch: mr.Source,
748 TargetBranch: mr.Target,
749 State: state,
750 Author: mr.Author,
751 CreatedAt: encodeTime(mr.CreatedAt),
752 UpdatedAt: encodeTime(mr.UpdatedAt),
753 HeadId: mr.HeadID.String(),
754 BaseId: mr.BaseID.String(),
755 MergedCommitId: mr.MergedCommitID.String(),
756 MergeMode: mode,
757 CommentSeq: mr.CommentSeq,
758 AutoMerge: autoMerge,
759 AutoMerged: mr.AutoMerged,
760 }), nil
761}
762
763func decodeMergeRequest(repo domain.RepoName, item kv.Item) (domain.MergeRequest, error) {
764 var pb storagepb.MergeRequest
765 if err := proto.Unmarshal(item.Value, &pb); err != nil {
766 return domain.MergeRequest{}, fmt.Errorf("decode merge request: %w", err)
767 }
768 fail := func(err error) (domain.MergeRequest, error) {
769 return domain.MergeRequest{}, fmt.Errorf("decode merge request %d of %q: %w", pb.GetNumber(), repo, err)
770 }
771 title, err := domain.ParseMRTitle(pb.GetTitle())
772 if err != nil {
773 return fail(err)
774 }
775 desc, err := domain.ParsePlainText(pb.GetDescription())
776 if err != nil {
777 return fail(err)
778 }
779 state, err := decodeState(pb.GetState())
780 if err != nil {
781 return fail(err)
782 }
783 mr := domain.MergeRequest{
784 Repo: repo, Number: pb.GetNumber(), Title: title, Description: desc,
785 Source: pb.GetSourceBranch(), Target: pb.GetTargetBranch(), State: state,
786 Author: pb.GetAuthor(),
787 CreatedAt: decodeTime(pb.GetCreatedAt()), UpdatedAt: decodeTime(pb.GetUpdatedAt()),
788 MergeMode: decodeMode(pb.GetMergeMode()),
789 CommentSeq: pb.GetCommentSeq(),
790 AutoMerged: pb.GetAutoMerged(),
791 Version: item.Version,
792 }
793 if am := pb.GetAutoMerge(); am != nil {
794 head, err := domain.ParseObjectID(am.GetHeadId())
795 if err != nil {
796 return fail(err)
797 }
798 mode := decodeMode(am.GetMode())
799 if mode == "" {
800 return fail(fmt.Errorf("unknown auto-merge mode %d", am.GetMode()))
801 }
802 mr.AutoMerge = &domain.AutoMerge{
803 Mode: mode, HeadID: head,
804 EnabledBy: domain.CIPusher{Kind: domain.CIPusherKind(am.GetEnablerKind()), TokenID: am.GetEnablerTokenId()},
805 EnablerLabel: am.GetEnablerLabel(), EnabledAt: decodeTime(am.GetEnabledAt()),
806 }
807 }
808 for _, f := range []struct {
809 dst *domain.ObjectID
810 src string
811 }{{&mr.HeadID, pb.GetHeadId()}, {&mr.BaseID, pb.GetBaseId()}, {&mr.MergedCommitID, pb.GetMergedCommitId()}} {
812 if f.src == "" {
813 continue
814 }
815 if *f.dst, err = domain.ParseObjectID(f.src); err != nil {
816 return fail(err)
817 }
818 }
819 return mr, nil
820}
821
822func encodeComment(c domain.Comment) ([]byte, error) {
823 side, err := encodeSide(c.Side)
824 if err != nil {
825 return nil, err
826 }
827 return mustMarshal(&storagepb.Comment{
828 ReplyTo: c.ReplyTo,
829 StartLine: int32(c.StartLine), //nolint:gosec // validated against the diff
830 Verdict: storagepb.Verdict(c.Verdict),
831 EditedAt: encodeTime(c.EditedAt),
832 Deleted: c.Deleted,
833 Id: c.ID,
834 Author: c.Author,
835 Path: c.Path,
836 Line: int32(c.Line), //nolint:gosec // validated against the diff, far below 2^31
837 Side: side,
838 CommitId: c.CommitID.String(),
839 Text: c.Text.String(),
840 Resolved: c.Resolved,
841 CreatedAt: encodeTime(c.CreatedAt),
842 }), nil
843}
844
845func decodeComment(item kv.Item) (domain.Comment, error) {
846 var pb storagepb.Comment
847 if err := proto.Unmarshal(item.Value, &pb); err != nil {
848 return domain.Comment{}, fmt.Errorf("decode comment: %w", err)
849 }
850 text, err := domain.ParsePlainText(pb.GetText())
851 if err != nil {
852 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", pb.GetId(), err)
853 }
854 c := domain.Comment{
855 ReplyTo: pb.GetReplyTo(), StartLine: int(pb.GetStartLine()), Verdict: domain.Verdict(pb.GetVerdict()), EditedAt: decodeTime(pb.GetEditedAt()), Deleted: pb.GetDeleted(),
856 ID: pb.GetId(), Author: pb.GetAuthor(), Path: pb.GetPath(), Line: int(pb.GetLine()),
857 Side: decodeSide(pb.GetSide()), Text: text, Resolved: pb.GetResolved(),
858 CreatedAt: decodeTime(pb.GetCreatedAt()), Version: item.Version,
859 }
860 if pb.GetCommitId() != "" {
861 if c.CommitID, err = domain.ParseObjectID(pb.GetCommitId()); err != nil {
862 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", c.ID, err)
863 }
864 }
865 return c, nil
866}
867
868func encodeState(s domain.MRState) (storagepb.MergeRequestState, error) {
869 switch s {
870 case domain.MRStateOpen:
871 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN, nil
872 case domain.MRStateMerged:
873 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED, nil
874 case domain.MRStateClosed:
875 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED, nil
876 }
877 return 0, fmt.Errorf("%w: merge request state %q", domain.ErrInvalid, s)
878}
879
880func decodeState(s storagepb.MergeRequestState) (domain.MRState, error) {
881 switch s {
882 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN:
883 return domain.MRStateOpen, nil
884 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED:
885 return domain.MRStateMerged, nil
886 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED:
887 return domain.MRStateClosed, nil
888 }
889 return "", fmt.Errorf("unknown merge request state %d", s)
890}
891
892func encodeMode(m domain.MergeMode) (storagepb.MergeMode, error) {
893 switch m {
894 case "":
895 return storagepb.MergeMode_MERGE_MODE_UNSPECIFIED, nil
896 case domain.MergeModeFastForward:
897 return storagepb.MergeMode_MERGE_MODE_FAST_FORWARD, nil
898 case domain.MergeModeSquash:
899 return storagepb.MergeMode_MERGE_MODE_SQUASH, nil
900 case domain.MergeModeMerge:
901 return storagepb.MergeMode_MERGE_MODE_MERGE, nil
902 }
903 return 0, fmt.Errorf("%w: merge mode %q", domain.ErrInvalid, m)
904}
905
906// decodeMode maps an unknown mode to none: it only labels a merged request.
907func decodeMode(m storagepb.MergeMode) domain.MergeMode {
908 switch m {
909 case storagepb.MergeMode_MERGE_MODE_FAST_FORWARD:
910 return domain.MergeModeFastForward
911 case storagepb.MergeMode_MERGE_MODE_SQUASH:
912 return domain.MergeModeSquash
913 case storagepb.MergeMode_MERGE_MODE_MERGE:
914 return domain.MergeModeMerge
915 }
916 return ""
917}
918
919func encodeSide(s domain.DiffSide) (storagepb.DiffSide, error) {
920 switch s {
921 case domain.SideNone:
922 return storagepb.DiffSide_DIFF_SIDE_UNSPECIFIED, nil
923 case domain.SideOld:
924 return storagepb.DiffSide_DIFF_SIDE_OLD, nil
925 case domain.SideNew:
926 return storagepb.DiffSide_DIFF_SIDE_NEW, nil
927 }
928 return 0, fmt.Errorf("%w: diff side %q", domain.ErrInvalid, s)
929}
930
931func decodeSide(s storagepb.DiffSide) domain.DiffSide {
932 switch s {
933 case storagepb.DiffSide_DIFF_SIDE_OLD:
934 return domain.SideOld
935 case storagepb.DiffSide_DIFF_SIDE_NEW:
936 return domain.SideNew
937 }
938 return domain.SideNone
939}
940
941func encodeSettings(s domain.RepoSettings) (*storagepb.RepoSettings, error) {
942 pb := &storagepb.RepoSettings{FailingChecksBlock: s.FailingChecksBlock, MergeModesV2: true, AllowAutoMerge: s.AllowAutoMerge}
943 for _, m := range s.AllowedMergeModes {
944 mode, err := encodeMode(m)
945 if err != nil || mode == storagepb.MergeMode_MERGE_MODE_UNSPECIFIED {
946 return nil, fmt.Errorf("%w: merge mode %q in the settings", domain.ErrInvalid, m)
947 }
948 pb.AllowedMergeModes = append(pb.AllowedMergeModes, mode)
949 }
950 issues, err := encodeIssuesMode(s.Issues)
951 if err != nil {
952 return nil, err
953 }
954 pb.Issues = issues
955 return pb, nil
956}
957
958// decodeSettings returns the defaults for a repository stored before merge
959// requests existed, and for settings left with no mode that this build knows.
960func decodeSettings(pb *storagepb.RepoSettings) domain.RepoSettings {
961 if pb == nil {
962 return domain.DefaultRepoSettings()
963 }
964 var modes []domain.MergeMode
965 for _, m := range pb.GetAllowedMergeModes() {
966 if mode := decodeMode(m); mode != "" {
967 modes = append(modes, mode)
968 }
969 }
970 if len(modes) == 0 {
971 return domain.DefaultRepoSettings()
972 }
973 normalized, err := domain.NormalizeMergeModes(modes)
974 if err != nil {
975 return domain.DefaultRepoSettings()
976 }
977 // Settings stored before merge commits: exactly {fast-forward, squash} was
978 // the old default, "everything allowed".
979 if !pb.GetMergeModesV2() && slices.Equal(normalized, []domain.MergeMode{domain.MergeModeFastForward, domain.MergeModeSquash}) {
980 normalized = domain.MergeModes()
981 }
982 return domain.RepoSettings{AllowedMergeModes: normalized, FailingChecksBlock: pb.GetFailingChecksBlock(), Issues: decodeIssuesMode(pb.GetIssues()), AllowAutoMerge: pb.GetAllowAutoMerge()}
983}