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