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.
99func indexOpen(b *kv.Batch, old *domain.MergeRequest, mr domain.MergeRequest) {
100 if old != nil && old.IsOpen() {
101 if !mr.IsOpen() || old.Source != mr.Source {
102 b.Delete(indexKey(mrOpenPrefix, old.Repo, old.Source, old.Number))
103 }
104 if !mr.IsOpen() || old.Target != mr.Target {
105 b.Delete(indexKey(mrTargetPrefix, old.Repo, old.Target, old.Number))
106 }
107 }
108 switch {
109 case mr.IsOpen():
110 b.Put(indexKey(mrOpenPrefix, mr.Repo, mr.Source, mr.Number), []byte{1})
111 b.Put(indexKey(mrTargetPrefix, mr.Repo, mr.Target, mr.Number), []byte{1})
112 case old == nil:
113 b.Delete(indexKey(mrOpenPrefix, mr.Repo, mr.Source, mr.Number))
114 b.Delete(indexKey(mrTargetPrefix, mr.Repo, mr.Target, mr.Number))
115 }
116}
117
118// BuildOpenIndex builds the indexes of the open merge requests (by source and
119// by target) when the data does not have them yet (it predates them): one pass
120// over the merge requests, in batches, then the marker. It runs when the store is opened and is idempotent: a pass
121// cut short by a crash is simply done again, writing the same keys.
122//
123//hexcheck:ignore suspicious-business-logic-in-adapter one-time storage backfill of an index: a storage concern, no policy
124func (s *Store) BuildOpenIndex(ctx context.Context) error {
125 if _, ok := s.db.Get(ctx, []byte(mrOpenIndexMarker)); ok {
126 return nil
127 }
128 var b kv.Batch
129 n := 0
130 flush := func() error {
131 if b.Len() == 0 {
132 return nil
133 }
134 if _, err := s.db.Apply(ctx, &b); err != nil {
135 return fmt.Errorf("build the index of open merge requests: %w", err)
136 }
137 b.Reset()
138 return nil
139 }
140 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(mrPrefix)) {
141 repoName, _, ok := strings.Cut(strings.TrimPrefix(string(key), mrPrefix), "/")
142 if !ok {
143 continue
144 }
145 repo, err := domain.ParseRepoName(repoName)
146 if err != nil {
147 continue
148 }
149 mr, err := decodeMergeRequest(repo, item)
150 if err != nil {
151 return fmt.Errorf("build the index of open merge requests: decode %s: %w", key, err)
152 }
153 if !mr.IsOpen() {
154 continue
155 }
156 b.Put(indexKey(mrOpenPrefix, mr.Repo, mr.Source, mr.Number), []byte{1})
157 b.Put(indexKey(mrTargetPrefix, mr.Repo, mr.Target, mr.Number), []byte{1})
158 if n++; n%(deleteChunk/2) == 0 {
159 if err := flush(); err != nil {
160 return err
161 }
162 }
163 }
164 if err := flush(); err != nil {
165 return err
166 }
167 b.Put([]byte(mrOpenIndexMarker), []byte{1})
168 if _, err := s.db.Apply(ctx, &b); err != nil {
169 return fmt.Errorf("build the index of open merge requests: %w", err)
170 }
171 log := zerowrap.FromCtx(ctx)
172 log.Info().Int("open", n).Msg("index of open merge requests built")
173 return nil
174}
175
176// mergeRequestKeys lists every key of the snapshot that belongs to the
177// repository's merge requests.
178func mergeRequestKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
179 var keys [][]byte
180 for _, p := range []string{
181 mrRepoPrefix(repo), commentPrefix + repo.String() + "/", readPrefix + repo.String() + "/",
182 mrOpenRepoPrefix(repo), mrTargetPrefix + repo.String() + "/",
183 mergePrefix + repo.String() + "/",
184 } {
185 for k := range snap.Prefix(ctx, []byte(p)) {
186 keys = append(keys, slices.Clone(k))
187 }
188 }
189 if _, ok := snap.Get(ctx, seqKey(repo)); ok {
190 keys = append(keys, seqKey(repo))
191 }
192 return keys
193}
194
195// guardRepo is guardRepoKey on the database of the merge requests; see there.
196func (m *MergeRequests) guardRepo(ctx context.Context, b *kv.Batch, repo domain.RepoName) error {
197 return guardRepoKey(ctx, m.db, b, repo)
198}
199
200// Create takes the next number of the repository from mrseq/<repo> and stores
201// the merge request under it, in one batch that requires the sequence to be
202// unchanged since it was read: two concurrent creations never share a number.
203// The batch also requires the repository to be unchanged since it was read and
204// rewrites it (see guardRepo), so that a repository deleted meanwhile is not
205// left with a stray merge request (a push recording pushed_at makes it retry). A missing repository gives
206// domain.ErrNotFound.
207//
208//hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
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 port method name only
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.
567//
568//hexcheck:ignore suspicious-business-logic-in-adapter retries a write that lost a race with the repository key: a storage concern, no policy
569func (m *MergeRequests) SetRead(ctx context.Context, repo domain.RepoName, number int64, head domain.ObjectID) error {
570 if head.IsZero() {
571 return fmt.Errorf("%w: empty read position", domain.ErrInvalidObjectID)
572 }
573 for range createTries {
574 var b kv.Batch
575 if err := m.guardRepo(ctx, &b, repo); err != nil {
576 return err
577 }
578 b.Put(readKey(repo, number), []byte(head.String()))
579 _, err := m.db.Apply(ctx, &b)
580 if errors.Is(err, kv.ErrConflict) {
581 continue
582 }
583 if err != nil {
584 return fmt.Errorf("record read position of %d in %q: %w", number, repo, err)
585 }
586 return nil
587 }
588 return fmt.Errorf("record read position of %d in %q: %w", number, repo, domain.ErrConflict)
589}
590
591// GetMergeability returns the stored mergeability, or domain.ErrNotFound.
592func (m *MergeRequests) GetMergeability(ctx context.Context, repo domain.RepoName, number int64) (domain.Mergeability, error) {
593 item, ok := m.db.Get(ctx, mergeKey(repo, number))
594 if !ok {
595 return domain.Mergeability{}, domain.ErrNotFound
596 }
597 return decodeMergeability(repo, number, item)
598}
599
600// PutMergeability stores the mergeability in a batch that requires the merge
601// request to be unchanged since it was read here and open, and the record to
602// be absent (Version 0) or unchanged.
603//
604//hexcheck:ignore suspicious-business-logic-in-adapter version preconditions of a derived record: a storage concern, no policy
605func (m *MergeRequests) PutMergeability(ctx context.Context, mg domain.Mergeability) error {
606 val, err := encodeMergeability(mg)
607 if err != nil {
608 return err
609 }
610 for range createTries {
611 mrItem, ok := m.db.Get(ctx, mrKey(mg.Repo, mg.Number))
612 if !ok {
613 return domain.ErrNotFound
614 }
615 mr, err := decodeMergeRequest(mg.Repo, mrItem)
616 if err != nil {
617 return err
618 }
619 if !mr.IsOpen() {
620 return domain.ErrNotFound
621 }
622 key := mergeKey(mg.Repo, mg.Number)
623 var b kv.Batch
624 b.RequireVersion(mrKey(mg.Repo, mg.Number), mrItem.Version)
625 if mg.Version == 0 {
626 b.RequireAbsent(key)
627 } else {
628 b.RequireVersion(key, mg.Version)
629 }
630 b.Put(key, val)
631 _, err = m.db.Apply(ctx, &b)
632 if errors.Is(err, kv.ErrConflict) {
633 // Either the record changed (the caller's conflict) or the merge
634 // request did (retry against it).
635 if cur, ok := m.db.Get(ctx, mrKey(mg.Repo, mg.Number)); ok && cur.Version != mrItem.Version {
636 continue
637 }
638 return domain.ErrConflict
639 }
640 if err != nil {
641 return fmt.Errorf("store mergeability of %d in %q: %w", mg.Number, mg.Repo, err)
642 }
643 return nil
644 }
645 return fmt.Errorf("store mergeability of %d in %q: %w", mg.Number, mg.Repo, domain.ErrConflict)
646}
647
648// DeleteMergeability removes the stored mergeability; absent is fine.
649func (m *MergeRequests) DeleteMergeability(ctx context.Context, repo domain.RepoName, number int64) error {
650 var b kv.Batch
651 b.Delete(mergeKey(repo, number))
652 if _, err := m.db.Apply(ctx, &b); err != nil {
653 return fmt.Errorf("delete mergeability of %d in %q: %w", number, repo, err)
654 }
655 return nil
656}
657
658func encodeMergeability(m domain.Mergeability) ([]byte, error) {
659 var state storagepb.MergeabilityState
660 switch m.State {
661 case domain.MergeabilityUpToDate:
662 state = storagepb.MergeabilityState_MERGEABILITY_STATE_UP_TO_DATE
663 case domain.MergeabilityBehind:
664 state = storagepb.MergeabilityState_MERGEABILITY_STATE_BEHIND
665 case domain.MergeabilityConflict:
666 state = storagepb.MergeabilityState_MERGEABILITY_STATE_CONFLICT
667 default:
668 return nil, fmt.Errorf("%w: mergeability state %q is not stored", domain.ErrInvalid, m.State)
669 }
670 return proto.Marshal(&storagepb.Mergeability{
671 State: state,
672 TargetId: m.TargetID.String(),
673 HeadId: m.HeadID.String(),
674 Behind: int32(m.Behind), //nolint:gosec // capped at MaxBehindCount
675 BehindCapped: m.BehindCapped,
676 ConflictPaths: m.ConflictPaths,
677 ConflictPathsTruncated: m.ConflictPathsTruncated,
678 CheckedAt: timestamppb.New(m.CheckedAt),
679 })
680}
681
682func decodeMergeability(repo domain.RepoName, number int64, item kv.Item) (domain.Mergeability, error) {
683 var pb storagepb.Mergeability
684 if err := proto.Unmarshal(item.Value, &pb); err != nil {
685 return domain.Mergeability{}, fmt.Errorf("decode mergeability of %d in %q: %w", number, repo, err)
686 }
687 state, ok := mergeabilityStates[pb.GetState()]
688 if !ok {
689 return domain.Mergeability{}, fmt.Errorf("decode mergeability of %d in %q: unknown state %d: %w", number, repo, pb.GetState(), domain.ErrNotFound)
690 }
691 target, err := domain.ParseObjectID(pb.GetTargetId())
692 if err != nil {
693 return domain.Mergeability{}, fmt.Errorf("decode mergeability of %d in %q: %w", number, repo, err)
694 }
695 head, err := domain.ParseObjectID(pb.GetHeadId())
696 if err != nil {
697 return domain.Mergeability{}, fmt.Errorf("decode mergeability of %d in %q: %w", number, repo, err)
698 }
699 return domain.Mergeability{
700 Repo: repo, Number: number, State: state,
701 TargetID: target, HeadID: head,
702 Behind: int(pb.GetBehind()), BehindCapped: pb.GetBehindCapped(),
703 ConflictPaths: pb.GetConflictPaths(),
704 ConflictPathsTruncated: pb.GetConflictPathsTruncated(),
705 CheckedAt: pb.GetCheckedAt().AsTime(),
706 Version: item.Version,
707 }, nil
708}
709
710// mergeabilityStates maps the stored states to the domain ones; CHECKING is
711// never stored.
712var mergeabilityStates = map[storagepb.MergeabilityState]domain.MergeabilityState{
713 storagepb.MergeabilityState_MERGEABILITY_STATE_UP_TO_DATE: domain.MergeabilityUpToDate,
714 storagepb.MergeabilityState_MERGEABILITY_STATE_BEHIND: domain.MergeabilityBehind,
715 storagepb.MergeabilityState_MERGEABILITY_STATE_CONFLICT: domain.MergeabilityConflict,
716}
717
718func encodeMergeRequest(mr domain.MergeRequest) ([]byte, error) {
719 if mr.Repo.IsZero() {
720 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
721 }
722 if !utf8.ValidString(mr.Author) {
723 return nil, errors.New("merge request author is not valid UTF-8")
724 }
725 state, err := encodeState(mr.State)
726 if err != nil {
727 return nil, err
728 }
729 mode, err := encodeMode(mr.MergeMode)
730 if err != nil {
731 return nil, err
732 }
733 return mustMarshal(&storagepb.MergeRequest{
734 Number: mr.Number,
735 Title: mr.Title.String(),
736 Description: mr.Description.String(),
737 SourceBranch: mr.Source,
738 TargetBranch: mr.Target,
739 State: state,
740 Author: mr.Author,
741 CreatedAt: encodeTime(mr.CreatedAt),
742 UpdatedAt: encodeTime(mr.UpdatedAt),
743 HeadId: mr.HeadID.String(),
744 BaseId: mr.BaseID.String(),
745 MergedCommitId: mr.MergedCommitID.String(),
746 MergeMode: mode,
747 CommentSeq: mr.CommentSeq,
748 }), nil
749}
750
751//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding, not a policy
752func decodeMergeRequest(repo domain.RepoName, item kv.Item) (domain.MergeRequest, error) {
753 var pb storagepb.MergeRequest
754 if err := proto.Unmarshal(item.Value, &pb); err != nil {
755 return domain.MergeRequest{}, fmt.Errorf("decode merge request: %w", err)
756 }
757 fail := func(err error) (domain.MergeRequest, error) {
758 return domain.MergeRequest{}, fmt.Errorf("decode merge request %d of %q: %w", pb.GetNumber(), repo, err)
759 }
760 title, err := domain.ParseMRTitle(pb.GetTitle())
761 if err != nil {
762 return fail(err)
763 }
764 desc, err := domain.ParsePlainText(pb.GetDescription())
765 if err != nil {
766 return fail(err)
767 }
768 state, err := decodeState(pb.GetState())
769 if err != nil {
770 return fail(err)
771 }
772 mr := domain.MergeRequest{
773 Repo: repo, Number: pb.GetNumber(), Title: title, Description: desc,
774 Source: pb.GetSourceBranch(), Target: pb.GetTargetBranch(), State: state,
775 Author: pb.GetAuthor(),
776 CreatedAt: decodeTime(pb.GetCreatedAt()), UpdatedAt: decodeTime(pb.GetUpdatedAt()),
777 MergeMode: decodeMode(pb.GetMergeMode()),
778 CommentSeq: pb.GetCommentSeq(),
779 Version: item.Version,
780 }
781 for _, f := range []struct {
782 dst *domain.ObjectID
783 src string
784 }{{&mr.HeadID, pb.GetHeadId()}, {&mr.BaseID, pb.GetBaseId()}, {&mr.MergedCommitID, pb.GetMergedCommitId()}} {
785 if f.src == "" {
786 continue
787 }
788 if *f.dst, err = domain.ParseObjectID(f.src); err != nil {
789 return fail(err)
790 }
791 }
792 return mr, nil
793}
794
795func encodeComment(c domain.Comment) ([]byte, error) {
796 side, err := encodeSide(c.Side)
797 if err != nil {
798 return nil, err
799 }
800 return mustMarshal(&storagepb.Comment{
801 ReplyTo: c.ReplyTo,
802 StartLine: int32(c.StartLine), //nolint:gosec // validated against the diff
803 Verdict: storagepb.Verdict(c.Verdict),
804 EditedAt: encodeTime(c.EditedAt),
805 Deleted: c.Deleted,
806 Id: c.ID,
807 Author: c.Author,
808 Path: c.Path,
809 Line: int32(c.Line), //nolint:gosec // validated against the diff, far below 2^31
810 Side: side,
811 CommitId: c.CommitID.String(),
812 Text: c.Text.String(),
813 Resolved: c.Resolved,
814 CreatedAt: encodeTime(c.CreatedAt),
815 }), nil
816}
817
818//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding, not a policy
819func decodeComment(item kv.Item) (domain.Comment, error) {
820 var pb storagepb.Comment
821 if err := proto.Unmarshal(item.Value, &pb); err != nil {
822 return domain.Comment{}, fmt.Errorf("decode comment: %w", err)
823 }
824 text, err := domain.ParsePlainText(pb.GetText())
825 if err != nil {
826 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", pb.GetId(), err)
827 }
828 c := domain.Comment{
829 ReplyTo: pb.GetReplyTo(), StartLine: int(pb.GetStartLine()), Verdict: domain.Verdict(pb.GetVerdict()), EditedAt: decodeTime(pb.GetEditedAt()), Deleted: pb.GetDeleted(),
830 ID: pb.GetId(), Author: pb.GetAuthor(), Path: pb.GetPath(), Line: int(pb.GetLine()),
831 Side: decodeSide(pb.GetSide()), Text: text, Resolved: pb.GetResolved(),
832 CreatedAt: decodeTime(pb.GetCreatedAt()), Version: item.Version,
833 }
834 if pb.GetCommitId() != "" {
835 if c.CommitID, err = domain.ParseObjectID(pb.GetCommitId()); err != nil {
836 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", c.ID, err)
837 }
838 }
839 return c, nil
840}
841
842func encodeState(s domain.MRState) (storagepb.MergeRequestState, error) {
843 switch s {
844 case domain.MRStateOpen:
845 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN, nil
846 case domain.MRStateMerged:
847 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED, nil
848 case domain.MRStateClosed:
849 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED, nil
850 }
851 return 0, fmt.Errorf("%w: merge request state %q", domain.ErrInvalid, s)
852}
853
854func decodeState(s storagepb.MergeRequestState) (domain.MRState, error) {
855 switch s {
856 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN:
857 return domain.MRStateOpen, nil
858 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED:
859 return domain.MRStateMerged, nil
860 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED:
861 return domain.MRStateClosed, nil
862 }
863 return "", fmt.Errorf("unknown merge request state %d", s)
864}
865
866func encodeMode(m domain.MergeMode) (storagepb.MergeMode, error) {
867 switch m {
868 case "":
869 return storagepb.MergeMode_MERGE_MODE_UNSPECIFIED, nil
870 case domain.MergeModeFastForward:
871 return storagepb.MergeMode_MERGE_MODE_FAST_FORWARD, nil
872 case domain.MergeModeSquash:
873 return storagepb.MergeMode_MERGE_MODE_SQUASH, nil
874 case domain.MergeModeMerge:
875 return storagepb.MergeMode_MERGE_MODE_MERGE, nil
876 }
877 return 0, fmt.Errorf("%w: merge mode %q", domain.ErrInvalid, m)
878}
879
880// decodeMode maps an unknown mode to none: it only labels a merged request.
881func decodeMode(m storagepb.MergeMode) domain.MergeMode {
882 switch m {
883 case storagepb.MergeMode_MERGE_MODE_FAST_FORWARD:
884 return domain.MergeModeFastForward
885 case storagepb.MergeMode_MERGE_MODE_SQUASH:
886 return domain.MergeModeSquash
887 case storagepb.MergeMode_MERGE_MODE_MERGE:
888 return domain.MergeModeMerge
889 }
890 return ""
891}
892
893func encodeSide(s domain.DiffSide) (storagepb.DiffSide, error) {
894 switch s {
895 case domain.SideNone:
896 return storagepb.DiffSide_DIFF_SIDE_UNSPECIFIED, nil
897 case domain.SideOld:
898 return storagepb.DiffSide_DIFF_SIDE_OLD, nil
899 case domain.SideNew:
900 return storagepb.DiffSide_DIFF_SIDE_NEW, nil
901 }
902 return 0, fmt.Errorf("%w: diff side %q", domain.ErrInvalid, s)
903}
904
905func decodeSide(s storagepb.DiffSide) domain.DiffSide {
906 switch s {
907 case storagepb.DiffSide_DIFF_SIDE_OLD:
908 return domain.SideOld
909 case storagepb.DiffSide_DIFF_SIDE_NEW:
910 return domain.SideNew
911 }
912 return domain.SideNone
913}
914
915func encodeSettings(s domain.RepoSettings) (*storagepb.RepoSettings, error) {
916 pb := &storagepb.RepoSettings{FailingChecksBlock: s.FailingChecksBlock, MergeModesV2: true}
917 for _, m := range s.AllowedMergeModes {
918 mode, err := encodeMode(m)
919 if err != nil || mode == storagepb.MergeMode_MERGE_MODE_UNSPECIFIED {
920 return nil, fmt.Errorf("%w: merge mode %q in the settings", domain.ErrInvalid, m)
921 }
922 pb.AllowedMergeModes = append(pb.AllowedMergeModes, mode)
923 }
924 issues, err := encodeIssuesMode(s.Issues)
925 if err != nil {
926 return nil, err
927 }
928 pb.Issues = issues
929 return pb, nil
930}
931
932// decodeSettings returns the defaults for a repository stored before merge
933// requests existed, and for settings left with no mode that this build knows.
934//
935//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding with the documented defaults
936func decodeSettings(pb *storagepb.RepoSettings) domain.RepoSettings {
937 if pb == nil {
938 return domain.DefaultRepoSettings()
939 }
940 var modes []domain.MergeMode
941 for _, m := range pb.GetAllowedMergeModes() {
942 if mode := decodeMode(m); mode != "" {
943 modes = append(modes, mode)
944 }
945 }
946 if len(modes) == 0 {
947 return domain.DefaultRepoSettings()
948 }
949 normalized, err := domain.NormalizeMergeModes(modes)
950 if err != nil {
951 return domain.DefaultRepoSettings()
952 }
953 // Settings stored before merge commits: exactly {fast-forward, squash} was
954 // the old default, "everything allowed".
955 if !pb.GetMergeModesV2() && slices.Equal(normalized, []domain.MergeMode{domain.MergeModeFastForward, domain.MergeModeSquash}) {
956 normalized = domain.MergeModes()
957 }
958 return domain.RepoSettings{AllowedMergeModes: normalized, FailingChecksBlock: pb.GetFailingChecksBlock(), Issues: decodeIssuesMode(pb.GetIssues())}
959}