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 return mustMarshal(&storagepb.MergeRequest{
732 Number: mr.Number,
733 Title: mr.Title.String(),
734 Description: mr.Description.String(),
735 SourceBranch: mr.Source,
736 TargetBranch: mr.Target,
737 State: state,
738 Author: mr.Author,
739 CreatedAt: encodeTime(mr.CreatedAt),
740 UpdatedAt: encodeTime(mr.UpdatedAt),
741 HeadId: mr.HeadID.String(),
742 BaseId: mr.BaseID.String(),
743 MergedCommitId: mr.MergedCommitID.String(),
744 MergeMode: mode,
745 CommentSeq: mr.CommentSeq,
746 }), nil
747}
748
749func decodeMergeRequest(repo domain.RepoName, item kv.Item) (domain.MergeRequest, error) {
750 var pb storagepb.MergeRequest
751 if err := proto.Unmarshal(item.Value, &pb); err != nil {
752 return domain.MergeRequest{}, fmt.Errorf("decode merge request: %w", err)
753 }
754 fail := func(err error) (domain.MergeRequest, error) {
755 return domain.MergeRequest{}, fmt.Errorf("decode merge request %d of %q: %w", pb.GetNumber(), repo, err)
756 }
757 title, err := domain.ParseMRTitle(pb.GetTitle())
758 if err != nil {
759 return fail(err)
760 }
761 desc, err := domain.ParsePlainText(pb.GetDescription())
762 if err != nil {
763 return fail(err)
764 }
765 state, err := decodeState(pb.GetState())
766 if err != nil {
767 return fail(err)
768 }
769 mr := domain.MergeRequest{
770 Repo: repo, Number: pb.GetNumber(), Title: title, Description: desc,
771 Source: pb.GetSourceBranch(), Target: pb.GetTargetBranch(), State: state,
772 Author: pb.GetAuthor(),
773 CreatedAt: decodeTime(pb.GetCreatedAt()), UpdatedAt: decodeTime(pb.GetUpdatedAt()),
774 MergeMode: decodeMode(pb.GetMergeMode()),
775 CommentSeq: pb.GetCommentSeq(),
776 Version: item.Version,
777 }
778 for _, f := range []struct {
779 dst *domain.ObjectID
780 src string
781 }{{&mr.HeadID, pb.GetHeadId()}, {&mr.BaseID, pb.GetBaseId()}, {&mr.MergedCommitID, pb.GetMergedCommitId()}} {
782 if f.src == "" {
783 continue
784 }
785 if *f.dst, err = domain.ParseObjectID(f.src); err != nil {
786 return fail(err)
787 }
788 }
789 return mr, nil
790}
791
792func encodeComment(c domain.Comment) ([]byte, error) {
793 side, err := encodeSide(c.Side)
794 if err != nil {
795 return nil, err
796 }
797 return mustMarshal(&storagepb.Comment{
798 ReplyTo: c.ReplyTo,
799 StartLine: int32(c.StartLine), //nolint:gosec // validated against the diff
800 Verdict: storagepb.Verdict(c.Verdict),
801 EditedAt: encodeTime(c.EditedAt),
802 Deleted: c.Deleted,
803 Id: c.ID,
804 Author: c.Author,
805 Path: c.Path,
806 Line: int32(c.Line), //nolint:gosec // validated against the diff, far below 2^31
807 Side: side,
808 CommitId: c.CommitID.String(),
809 Text: c.Text.String(),
810 Resolved: c.Resolved,
811 CreatedAt: encodeTime(c.CreatedAt),
812 }), nil
813}
814
815func decodeComment(item kv.Item) (domain.Comment, error) {
816 var pb storagepb.Comment
817 if err := proto.Unmarshal(item.Value, &pb); err != nil {
818 return domain.Comment{}, fmt.Errorf("decode comment: %w", err)
819 }
820 text, err := domain.ParsePlainText(pb.GetText())
821 if err != nil {
822 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", pb.GetId(), err)
823 }
824 c := domain.Comment{
825 ReplyTo: pb.GetReplyTo(), StartLine: int(pb.GetStartLine()), Verdict: domain.Verdict(pb.GetVerdict()), EditedAt: decodeTime(pb.GetEditedAt()), Deleted: pb.GetDeleted(),
826 ID: pb.GetId(), Author: pb.GetAuthor(), Path: pb.GetPath(), Line: int(pb.GetLine()),
827 Side: decodeSide(pb.GetSide()), Text: text, Resolved: pb.GetResolved(),
828 CreatedAt: decodeTime(pb.GetCreatedAt()), Version: item.Version,
829 }
830 if pb.GetCommitId() != "" {
831 if c.CommitID, err = domain.ParseObjectID(pb.GetCommitId()); err != nil {
832 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", c.ID, err)
833 }
834 }
835 return c, nil
836}
837
838func encodeState(s domain.MRState) (storagepb.MergeRequestState, error) {
839 switch s {
840 case domain.MRStateOpen:
841 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN, nil
842 case domain.MRStateMerged:
843 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED, nil
844 case domain.MRStateClosed:
845 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED, nil
846 }
847 return 0, fmt.Errorf("%w: merge request state %q", domain.ErrInvalid, s)
848}
849
850func decodeState(s storagepb.MergeRequestState) (domain.MRState, error) {
851 switch s {
852 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN:
853 return domain.MRStateOpen, nil
854 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED:
855 return domain.MRStateMerged, nil
856 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED:
857 return domain.MRStateClosed, nil
858 }
859 return "", fmt.Errorf("unknown merge request state %d", s)
860}
861
862func encodeMode(m domain.MergeMode) (storagepb.MergeMode, error) {
863 switch m {
864 case "":
865 return storagepb.MergeMode_MERGE_MODE_UNSPECIFIED, nil
866 case domain.MergeModeFastForward:
867 return storagepb.MergeMode_MERGE_MODE_FAST_FORWARD, nil
868 case domain.MergeModeSquash:
869 return storagepb.MergeMode_MERGE_MODE_SQUASH, nil
870 case domain.MergeModeMerge:
871 return storagepb.MergeMode_MERGE_MODE_MERGE, nil
872 }
873 return 0, fmt.Errorf("%w: merge mode %q", domain.ErrInvalid, m)
874}
875
876// decodeMode maps an unknown mode to none: it only labels a merged request.
877func decodeMode(m storagepb.MergeMode) domain.MergeMode {
878 switch m {
879 case storagepb.MergeMode_MERGE_MODE_FAST_FORWARD:
880 return domain.MergeModeFastForward
881 case storagepb.MergeMode_MERGE_MODE_SQUASH:
882 return domain.MergeModeSquash
883 case storagepb.MergeMode_MERGE_MODE_MERGE:
884 return domain.MergeModeMerge
885 }
886 return ""
887}
888
889func encodeSide(s domain.DiffSide) (storagepb.DiffSide, error) {
890 switch s {
891 case domain.SideNone:
892 return storagepb.DiffSide_DIFF_SIDE_UNSPECIFIED, nil
893 case domain.SideOld:
894 return storagepb.DiffSide_DIFF_SIDE_OLD, nil
895 case domain.SideNew:
896 return storagepb.DiffSide_DIFF_SIDE_NEW, nil
897 }
898 return 0, fmt.Errorf("%w: diff side %q", domain.ErrInvalid, s)
899}
900
901func decodeSide(s storagepb.DiffSide) domain.DiffSide {
902 switch s {
903 case storagepb.DiffSide_DIFF_SIDE_OLD:
904 return domain.SideOld
905 case storagepb.DiffSide_DIFF_SIDE_NEW:
906 return domain.SideNew
907 }
908 return domain.SideNone
909}
910
911func encodeSettings(s domain.RepoSettings) (*storagepb.RepoSettings, error) {
912 pb := &storagepb.RepoSettings{FailingChecksBlock: s.FailingChecksBlock, MergeModesV2: true}
913 for _, m := range s.AllowedMergeModes {
914 mode, err := encodeMode(m)
915 if err != nil || mode == storagepb.MergeMode_MERGE_MODE_UNSPECIFIED {
916 return nil, fmt.Errorf("%w: merge mode %q in the settings", domain.ErrInvalid, m)
917 }
918 pb.AllowedMergeModes = append(pb.AllowedMergeModes, mode)
919 }
920 issues, err := encodeIssuesMode(s.Issues)
921 if err != nil {
922 return nil, err
923 }
924 pb.Issues = issues
925 return pb, nil
926}
927
928// decodeSettings returns the defaults for a repository stored before merge
929// requests existed, and for settings left with no mode that this build knows.
930func decodeSettings(pb *storagepb.RepoSettings) domain.RepoSettings {
931 if pb == nil {
932 return domain.DefaultRepoSettings()
933 }
934 var modes []domain.MergeMode
935 for _, m := range pb.GetAllowedMergeModes() {
936 if mode := decodeMode(m); mode != "" {
937 modes = append(modes, mode)
938 }
939 }
940 if len(modes) == 0 {
941 return domain.DefaultRepoSettings()
942 }
943 normalized, err := domain.NormalizeMergeModes(modes)
944 if err != nil {
945 return domain.DefaultRepoSettings()
946 }
947 // Settings stored before merge commits: exactly {fast-forward, squash} was
948 // the old default, "everything allowed".
949 if !pb.GetMergeModesV2() && slices.Equal(normalized, []domain.MergeMode{domain.MergeModeFastForward, domain.MergeModeSquash}) {
950 normalized = domain.MergeModes()
951 }
952 return domain.RepoSettings{AllowedMergeModes: normalized, FailingChecksBlock: pb.GetFailingChecksBlock(), Issues: decodeIssuesMode(pb.GetIssues())}
953}